diff --git a/CHANGES.txt b/CHANGES.txt index 78518b68d9..4f7d6ddea0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.6 + * Add uncompressed chunk cache for RandomAccessReader (CASSANDRA-5863) * Clarify ClusteringPrefix hierarchy (CASSANDRA-11213) * Always perform collision check before joining ring (CASSANDRA-10134) * SSTableWriter output discrepancy (CASSANDRA-11646) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 582859c024..48bad2c728 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -380,9 +380,14 @@ concurrent_counter_writes: 32 # be limited by the less of concurrent reads or concurrent writes. concurrent_materialized_view_writes: 32 -# 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. +# Maximum memory to use for sstable chunk cache and buffer pooling. +# 32MB of this are reserved for pooling buffers, the rest is used as an +# cache that holds uncompressed sstable chunks. +# 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. The cache also has on-heap +# overhead which is roughly 128 bytes per chunk (i.e. 0.2% of the reserved size +# if the default 64k chunk size is used). +# Memory is only allocated when needed. # file_cache_size_in_mb: 512 # Flag indicating whether to allocate on or off heap when the sstable buffer diff --git a/lib/caffeine-2.2.6.jar b/lib/caffeine-2.2.6.jar new file mode 100644 index 0000000000..74b91bc749 Binary files /dev/null and b/lib/caffeine-2.2.6.jar differ diff --git a/lib/licenses/caffeine-2.2.6.txt b/lib/licenses/caffeine-2.2.6.txt new file mode 100644 index 0000000000..d645695673 --- /dev/null +++ b/lib/licenses/caffeine-2.2.6.txt @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed 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. diff --git a/src/java/org/apache/cassandra/cache/CacheSize.java b/src/java/org/apache/cassandra/cache/CacheSize.java new file mode 100644 index 0000000000..561c73dbf1 --- /dev/null +++ b/src/java/org/apache/cassandra/cache/CacheSize.java @@ -0,0 +1,14 @@ +package org.apache.cassandra.cache; + +public interface CacheSize +{ + + long capacity(); + + void setCapacity(long capacity); + + int size(); + + long weightedSize(); + +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/cache/ChunkCache.java b/src/java/org/apache/cassandra/cache/ChunkCache.java new file mode 100644 index 0000000000..faf41b4be0 --- /dev/null +++ b/src/java/org/apache/cassandra/cache/ChunkCache.java @@ -0,0 +1,301 @@ +package org.apache.cassandra.cache; + +import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicInteger; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Throwables; +import com.google.common.collect.Iterables; +import com.google.common.util.concurrent.MoreExecutors; + +import com.github.benmanes.caffeine.cache.*; +import com.codahale.metrics.Timer; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.io.sstable.CorruptSSTableException; +import org.apache.cassandra.io.util.*; +import org.apache.cassandra.metrics.CacheMissMetrics; +import org.apache.cassandra.utils.memory.BufferPool; + +public class ChunkCache + implements CacheLoader, RemovalListener, CacheSize +{ + public static final int RESERVED_POOL_SPACE_IN_MB = 32; + public static final long cacheSize = 1024L * 1024L * Math.max(0, DatabaseDescriptor.getFileCacheSizeInMB() - RESERVED_POOL_SPACE_IN_MB); + + private static boolean enabled = cacheSize > 0; + public static final ChunkCache instance = enabled ? new ChunkCache() : null; + + private final LoadingCache cache; + public final CacheMissMetrics metrics; + + static class Key + { + final ChunkReader file; + final String path; + final long position; + + public Key(ChunkReader file, long position) + { + super(); + this.file = file; + this.position = position; + this.path = file.channel().filePath(); + } + + public int hashCode() + { + final int prime = 31; + int result = 1; + result = prime * result + path.hashCode(); + result = prime * result + file.getClass().hashCode(); + result = prime * result + Long.hashCode(position); + return result; + } + + public boolean equals(Object obj) + { + if (this == obj) + return true; + if (obj == null) + return false; + + Key other = (Key) obj; + return (position == other.position) + && file.getClass() == other.file.getClass() + && path.equals(other.path); + } + } + + static class Buffer implements Rebufferer.BufferHolder + { + private final ByteBuffer buffer; + private final long offset; + private final AtomicInteger references; + + public Buffer(ByteBuffer buffer, long offset) + { + this.buffer = buffer; + this.offset = offset; + references = new AtomicInteger(1); // start referenced. + } + + Buffer reference() + { + int refCount; + do + { + refCount = references.get(); + if (refCount == 0) + // Buffer was released before we managed to reference it. + return null; + } while (!references.compareAndSet(refCount, refCount + 1)); + + return this; + } + + @Override + public ByteBuffer buffer() + { + assert references.get() > 0; + return buffer.duplicate(); + } + + @Override + public long offset() + { + return offset; + } + + @Override + public void release() + { + if (references.decrementAndGet() == 0) + BufferPool.put(buffer); + } + } + + public ChunkCache() + { + cache = Caffeine.newBuilder() + .maximumWeight(cacheSize) + .executor(MoreExecutors.directExecutor()) + .weigher((key, buffer) -> ((Buffer) buffer).buffer.capacity()) + .removalListener(this) + .build(this); + metrics = new CacheMissMetrics("ChunkCache", this); + } + + @Override + public Buffer load(Key key) throws Exception + { + ChunkReader rebufferer = key.file; + metrics.misses.mark(); + try (Timer.Context ctx = metrics.missLatency.time()) + { + ByteBuffer buffer = BufferPool.get(key.file.chunkSize()); + assert buffer != null; + rebufferer.readChunk(key.position, buffer); + return new Buffer(buffer, key.position); + } + } + + @Override + public void onRemoval(Key key, Buffer buffer, RemovalCause cause) + { + buffer.release(); + } + + public void close() + { + cache.invalidateAll(); + } + + public RebuffererFactory wrap(ChunkReader file) + { + return new CachingRebufferer(file); + } + + public static RebuffererFactory maybeWrap(ChunkReader file) + { + if (!enabled) + return file; + + return instance.wrap(file); + } + + public void invalidatePosition(SegmentedFile dfile, long position) + { + if (!(dfile.rebuffererFactory() instanceof CachingRebufferer)) + return; + + ((CachingRebufferer) dfile.rebuffererFactory()).invalidate(position); + } + + public void invalidateFile(String fileName) + { + cache.invalidateAll(Iterables.filter(cache.asMap().keySet(), x -> x.path.equals(fileName))); + } + + @VisibleForTesting + public void enable(boolean enabled) + { + ChunkCache.enabled = enabled; + cache.invalidateAll(); + metrics.reset(); + } + + // TODO: Invalidate caches for obsoleted/MOVED_START tables? + + /** + * Rebufferer providing cached chunks where data is obtained from the specified ChunkReader. + * Thread-safe. One instance per SegmentedFile, created by ChunkCache.maybeWrap if the cache is enabled. + */ + class CachingRebufferer implements Rebufferer, RebuffererFactory + { + private final ChunkReader source; + final long alignmentMask; + + public CachingRebufferer(ChunkReader file) + { + source = file; + int chunkSize = file.chunkSize(); + assert Integer.bitCount(chunkSize) == 1; // Must be power of two + alignmentMask = -chunkSize; + } + + @Override + public Buffer rebuffer(long position) + { + try + { + metrics.requests.mark(); + long pageAlignedPos = position & alignmentMask; + Buffer buf; + do + buf = cache.get(new Key(source, pageAlignedPos)).reference(); + while (buf == null); + + return buf; + } + catch (Throwable t) + { + Throwables.propagateIfInstanceOf(t.getCause(), CorruptSSTableException.class); + throw Throwables.propagate(t); + } + } + + public void invalidate(long position) + { + long pageAlignedPos = position & alignmentMask; + cache.invalidate(new Key(source, pageAlignedPos)); + } + + @Override + public Rebufferer instantiateRebufferer() + { + return this; + } + + @Override + public void close() + { + source.close(); + } + + @Override + public void closeReader() + { + // Instance is shared among readers. Nothing to release. + } + + @Override + public ChannelProxy channel() + { + return source.channel(); + } + + @Override + public long fileLength() + { + return source.fileLength(); + } + + @Override + public double getCrcCheckChance() + { + return source.getCrcCheckChance(); + } + + @Override + public String toString() + { + return "CachingRebufferer:" + source.toString(); + } + } + + @Override + public long capacity() + { + return cacheSize; + } + + @Override + public void setCapacity(long capacity) + { + throw new UnsupportedOperationException("Chunk cache size cannot be changed."); + } + + @Override + public int size() + { + return cache.asMap().size(); + } + + @Override + public long weightedSize() + { + return cache.policy().eviction() + .map(policy -> policy.weightedSize().orElseGet(cache::estimatedSize)) + .orElseGet(cache::estimatedSize); + } +} diff --git a/src/java/org/apache/cassandra/cache/ICache.java b/src/java/org/apache/cassandra/cache/ICache.java index 37b55cd0e7..7ca6b2e763 100644 --- a/src/java/org/apache/cassandra/cache/ICache.java +++ b/src/java/org/apache/cassandra/cache/ICache.java @@ -24,12 +24,8 @@ import java.util.Iterator; * and does not require put or remove to return values, which lets SerializingCache * be more efficient by avoiding deserialize except on get. */ -public interface ICache +public interface ICache extends CacheSize { - public long capacity(); - - public void setCapacity(long capacity); - public void put(K key, V value); public boolean putIfAbsent(K key, V value); @@ -40,10 +36,6 @@ public interface ICache public void remove(K key); - public int size(); - - public long weightedSize(); - public void clear(); public Iterator keyIterator(); diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java index 67d351acd7..2d70da64a6 100644 --- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java +++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java @@ -443,7 +443,7 @@ public class Scrubber implements Closeable } catch (Exception e) { - throw new RuntimeException(); + throw new RuntimeException(e); } } } diff --git a/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java b/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java index 095d7f4109..8bb5b6d3d1 100644 --- a/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java +++ b/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java @@ -22,9 +22,11 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.zip.CRC32; -import org.apache.cassandra.io.util.ChannelProxy; -import org.apache.cassandra.io.util.DataPosition; -import org.apache.cassandra.io.util.RandomAccessReader; +import com.google.common.base.Preconditions; + +import org.apache.cassandra.io.compress.BufferType; +import org.apache.cassandra.io.util.*; +import org.apache.cassandra.utils.memory.BufferPool; /** * A {@link RandomAccessReader} wrapper that calctulates the CRC in place. @@ -37,35 +39,55 @@ import org.apache.cassandra.io.util.RandomAccessReader; * corrupted sequence by reading a huge corrupted length of bytes via * via {@link org.apache.cassandra.utils.ByteBufferUtil#readWithLength(java.io.DataInput)}. */ -public class ChecksummedDataInput extends RandomAccessReader.RandomAccessReaderWithOwnChannel +public class ChecksummedDataInput extends RebufferingInputStream { private final CRC32 crc; private int crcPosition; private boolean crcUpdateDisabled; private long limit; - private DataPosition limitMark; + private long limitMark; - protected ChecksummedDataInput(Builder builder) + protected long bufferOffset; + protected final ChannelProxy channel; + + ChecksummedDataInput(ChannelProxy channel, BufferType bufferType) { - super(builder); + super(BufferPool.get(RandomAccessReader.DEFAULT_BUFFER_SIZE, bufferType)); crc = new CRC32(); crcPosition = 0; crcUpdateDisabled = false; + this.channel = channel; + bufferOffset = 0; + buffer.limit(0); resetLimit(); } - @SuppressWarnings("resource") // channel owned by RandomAccessReaderWithOwnChannel - public static ChecksummedDataInput open(File file) + ChecksummedDataInput(ChannelProxy channel) { - return new Builder(new ChannelProxy(file)).build(); + this(channel, BufferType.OFF_HEAP); } - protected void releaseBuffer() + @SuppressWarnings("resource") + public static ChecksummedDataInput open(File file) { - super.releaseBuffer(); + return new ChecksummedDataInput(new ChannelProxy(file)); + } + + public boolean isEOF() + { + return getPosition() == channel.size(); + } + + /** + * Returns the position in the source file, which is different for getPosition() for compressed/encrypted files + * and may be imprecise. + */ + public long getSourcePosition() + { + return getPosition(); } public void resetCrc() @@ -76,29 +98,34 @@ public class ChecksummedDataInput extends RandomAccessReader.RandomAccessReaderW public void limit(long newLimit) { - limit = newLimit; - limitMark = mark(); + limitMark = getPosition(); + limit = limitMark + newLimit; + } + + /** + * Returns the exact position in the uncompressed view of the file. + */ + protected long getPosition() + { + return bufferOffset + buffer.position(); } public void resetLimit() { limit = Long.MAX_VALUE; - limitMark = null; + limitMark = -1; } public void checkLimit(int length) throws IOException { - if (limitMark == null) - return; - - if ((bytesPastLimit() + length) > limit) + if (getPosition() + length > limit) throw new IOException("Digest mismatch exception"); } public long bytesPastLimit() { - assert limitMark != null; - return bytesPastMark(limitMark); + assert limitMark != -1; + return getPosition() - limitMark; } public boolean checkCrc() throws IOException @@ -134,13 +161,24 @@ public class ChecksummedDataInput extends RandomAccessReader.RandomAccessReaderW } @Override - public void reBuffer() + protected void reBuffer() { + Preconditions.checkState(buffer.remaining() == 0); updateCrc(); - super.reBuffer(); + bufferOffset += buffer.limit(); + + readBuffer(); + crcPosition = buffer.position(); } + protected void readBuffer() + { + buffer.clear(); + while ((channel.read(buffer, bufferOffset)) == 0) {} + buffer.flip(); + } + private void updateCrc() { if (crcPosition == buffer.position() || crcUpdateDisabled) @@ -155,16 +193,20 @@ public class ChecksummedDataInput extends RandomAccessReader.RandomAccessReaderW crc.update(unprocessed); } - public static class Builder extends RandomAccessReader.Builder + @Override + public void close() { - public Builder(ChannelProxy channel) - { - super(channel); - } + BufferPool.put(buffer); + channel.close(); + } - public ChecksummedDataInput build() - { - return new ChecksummedDataInput(this); - } + protected String getPath() + { + return channel.filePath(); + } + + public ChannelProxy getChannel() + { + return channel; } } diff --git a/src/java/org/apache/cassandra/hints/CompressedChecksummedDataInput.java b/src/java/org/apache/cassandra/hints/CompressedChecksummedDataInput.java index c43e46e17f..f584dd10b2 100644 --- a/src/java/org/apache/cassandra/hints/CompressedChecksummedDataInput.java +++ b/src/java/org/apache/cassandra/hints/CompressedChecksummedDataInput.java @@ -35,13 +35,11 @@ public final class CompressedChecksummedDataInput extends ChecksummedDataInput private volatile ByteBuffer compressedBuffer = null; private final ByteBuffer metadataBuffer = ByteBuffer.allocate(CompressedHintsWriter.METADATA_SIZE); - public CompressedChecksummedDataInput(CompressedChecksummedDataInputBuilder builder) + public CompressedChecksummedDataInput(ChannelProxy channel, ICompressor compressor, long filePosition) { - super(builder); - assert regions == null; //mmapped regions are not supported - - compressor = builder.compressor; - filePosition = builder.position; + super(channel, compressor.preferredBufferType()); + this.compressor = compressor; + this.filePosition = filePosition; } /** @@ -53,7 +51,13 @@ public final class CompressedChecksummedDataInput extends ChecksummedDataInput return filePosition == channel.size() && buffer.remaining() == 0; } - protected void reBufferStandard() + public long getSourcePosition() + { + return filePosition; + } + + @Override + protected void readBuffer() { metadataBuffer.clear(); channel.read(metadataBuffer, filePosition); @@ -70,7 +74,7 @@ public final class CompressedChecksummedDataInput extends ChecksummedDataInput { BufferPool.put(compressedBuffer); } - compressedBuffer = allocateBuffer(bufferSize, compressor.preferredBufferType()); + compressedBuffer = BufferPool.get(bufferSize, compressor.preferredBufferType()); } compressedBuffer.clear(); @@ -79,12 +83,11 @@ public final class CompressedChecksummedDataInput extends ChecksummedDataInput compressedBuffer.rewind(); filePosition += compressedSize; - bufferOffset += buffer.position(); if (buffer.capacity() < uncompressedSize) { int bufferSize = uncompressedSize + (uncompressedSize / 20); BufferPool.put(buffer); - buffer = allocateBuffer(bufferSize, compressor.preferredBufferType()); + buffer = BufferPool.get(bufferSize, compressor.preferredBufferType()); } buffer.clear(); @@ -100,19 +103,11 @@ public final class CompressedChecksummedDataInput extends ChecksummedDataInput } } - protected void releaseBuffer() + @Override + public void close() { - super.releaseBuffer(); - if (compressedBuffer != null) - { - BufferPool.put(compressedBuffer); - compressedBuffer = null; - } - } - - protected void reBufferMmap() - { - throw new UnsupportedOperationException(); + BufferPool.put(compressedBuffer); + super.close(); } @SuppressWarnings("resource") // Closing the ChecksummedDataInput will close the underlying channel. @@ -121,10 +116,7 @@ public final class CompressedChecksummedDataInput extends ChecksummedDataInput long position = input.getPosition(); input.close(); - CompressedChecksummedDataInputBuilder builder = new CompressedChecksummedDataInputBuilder(new ChannelProxy(input.getPath())); - builder.withPosition(position); - builder.withCompressor(compressor); - return builder.build(); + return new CompressedChecksummedDataInput(new ChannelProxy(input.getPath()), compressor, position); } @VisibleForTesting diff --git a/src/java/org/apache/cassandra/hints/CompressedChecksummedDataInputBuilder.java b/src/java/org/apache/cassandra/hints/CompressedChecksummedDataInputBuilder.java deleted file mode 100644 index 3452df842b..0000000000 --- a/src/java/org/apache/cassandra/hints/CompressedChecksummedDataInputBuilder.java +++ /dev/null @@ -1,36 +0,0 @@ -package org.apache.cassandra.hints; - -import org.apache.cassandra.io.compress.ICompressor; -import org.apache.cassandra.io.util.ChannelProxy; - -public class CompressedChecksummedDataInputBuilder extends ChecksummedDataInput.Builder -{ - long position; - ICompressor compressor; - - public CompressedChecksummedDataInputBuilder(ChannelProxy channel) - { - super(channel); - bufferType = null; - } - - public ChecksummedDataInput build() - { - assert position >= 0; - assert compressor != null; - return new CompressedChecksummedDataInput(this); - } - - public CompressedChecksummedDataInputBuilder withCompressor(ICompressor compressor) - { - this.compressor = compressor; - bufferType = compressor.preferredBufferType(); - return this; - } - - public CompressedChecksummedDataInputBuilder withPosition(long position) - { - this.position = position; - return this; - } -} diff --git a/src/java/org/apache/cassandra/hints/EncryptedChecksummedDataInput.java b/src/java/org/apache/cassandra/hints/EncryptedChecksummedDataInput.java index 12b6bf24d4..7ecfbfe845 100644 --- a/src/java/org/apache/cassandra/hints/EncryptedChecksummedDataInput.java +++ b/src/java/org/apache/cassandra/hints/EncryptedChecksummedDataInput.java @@ -43,12 +43,12 @@ public class EncryptedChecksummedDataInput extends ChecksummedDataInput private final EncryptionUtils.ChannelProxyReadChannel readChannel; - protected EncryptedChecksummedDataInput(Builder builder) + protected EncryptedChecksummedDataInput(ChannelProxy channel, Cipher cipher, ICompressor compressor, long filePosition) { - super(builder); - cipher = builder.cipher; - compressor = builder.compressor; - readChannel = new EncryptionUtils.ChannelProxyReadChannel(channel, builder.position); + super(channel); + this.cipher = cipher; + this.compressor = compressor; + readChannel = new EncryptionUtils.ChannelProxyReadChannel(channel, filePosition); assert cipher != null; assert compressor != null; } @@ -59,10 +59,16 @@ public class EncryptedChecksummedDataInput extends ChecksummedDataInput */ public boolean isEOF() { - return readChannel.getCurrentPosition() == channel.size() && buffer.remaining() == 0; + return getSourcePosition() == channel.size() && buffer.remaining() == 0; } - protected void reBufferStandard() + public long getSourcePosition() + { + return readChannel.getCurrentPosition(); + } + + @Override + protected void readBuffer() { try { @@ -79,40 +85,13 @@ public class EncryptedChecksummedDataInput extends ChecksummedDataInput } } - public static class Builder extends CompressedChecksummedDataInputBuilder - { - Cipher cipher; - - public Builder(ChannelProxy channel) - { - super(channel); - } - - public Builder withCipher(Cipher cipher) - { - this.cipher = cipher; - return this; - } - - public ChecksummedDataInput build() - { - assert position >= 0; - assert compressor != null; - assert cipher != null; - return new EncryptedChecksummedDataInput(this); - } - } - + @SuppressWarnings("resource") public static ChecksummedDataInput upgradeInput(ChecksummedDataInput input, Cipher cipher, ICompressor compressor) { long position = input.getPosition(); input.close(); - Builder builder = new Builder(new ChannelProxy(input.getPath())); - builder.withPosition(position); - builder.withCompressor(compressor); - builder.withCipher(cipher); - return builder.build(); + return new EncryptedChecksummedDataInput(new ChannelProxy(input.getPath()), cipher, compressor, position); } @VisibleForTesting diff --git a/src/java/org/apache/cassandra/hints/HintsReader.java b/src/java/org/apache/cassandra/hints/HintsReader.java index 0571af4652..5e73805a01 100644 --- a/src/java/org/apache/cassandra/hints/HintsReader.java +++ b/src/java/org/apache/cassandra/hints/HintsReader.java @@ -111,7 +111,7 @@ class HintsReader implements AutoCloseable, Iterable void seek(long newPosition) { - input.seek(newPosition); + throw new UnsupportedOperationException("Hints are not seekable."); } public Iterator iterator() @@ -149,12 +149,12 @@ class HintsReader implements AutoCloseable, Iterable @SuppressWarnings("resource") protected Page computeNext() { - CLibrary.trySkipCache(input.getChannel().getFileDescriptor(), 0, input.getFilePointer(), input.getPath()); + CLibrary.trySkipCache(input.getChannel().getFileDescriptor(), 0, input.getSourcePosition(), input.getPath()); if (input.isEOF()) return endOfData(); - return new Page(input.getFilePointer()); + return new Page(input.getSourcePosition()); } } @@ -177,7 +177,7 @@ class HintsReader implements AutoCloseable, Iterable do { - long position = input.getFilePointer(); + long position = input.getSourcePosition(); if (input.isEOF()) return endOfData(); // reached EOF @@ -267,7 +267,7 @@ class HintsReader implements AutoCloseable, Iterable do { - long position = input.getFilePointer(); + long position = input.getSourcePosition(); if (input.isEOF()) return endOfData(); // reached EOF diff --git a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java deleted file mode 100644 index 329d93256e..0000000000 --- a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java +++ /dev/null @@ -1,286 +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.compress; - -import java.io.*; -import java.nio.ByteBuffer; -import java.util.concurrent.ThreadLocalRandom; -import java.util.zip.Checksum; -import java.util.function.Supplier; - -import com.google.common.annotations.VisibleForTesting; -import com.google.common.primitives.Ints; - -import org.apache.cassandra.io.FSReadError; -import org.apache.cassandra.io.sstable.CorruptSSTableException; -import org.apache.cassandra.io.util.*; -import org.apache.cassandra.utils.memory.BufferPool; - -/** - * CRAR extends RAR to transparently uncompress blocks from the file into RAR.buffer. Most of the RAR - * "read bytes from the buffer, rebuffering when necessary" machinery works unchanged after that. - */ -public class CompressedRandomAccessReader extends RandomAccessReader -{ - private final CompressionMetadata metadata; - - // we read the raw compressed bytes into this buffer, then move the uncompressed ones into super.buffer. - private ByteBuffer compressed; - - // re-use single crc object - private final Checksum checksum; - - // raw checksum bytes - private ByteBuffer checksumBytes; - - @VisibleForTesting - public double getCrcCheckChance() - { - return metadata.parameters.getCrcCheckChance(); - } - - protected CompressedRandomAccessReader(Builder builder) - { - super(builder); - this.metadata = builder.metadata; - this.checksum = metadata.checksumType.newInstance(); - - if (regions == null) - { - compressed = allocateBuffer(metadata.compressor().initialCompressedBufferLength(metadata.chunkLength()), bufferType); - checksumBytes = ByteBuffer.wrap(new byte[4]); - } - } - - @Override - protected void releaseBuffer() - { - try - { - if (buffer != null) - { - BufferPool.put(buffer); - buffer = null; - } - } - finally - { - // this will always be null if using mmap access mode (unlike in parent, where buffer is set to a region) - if (compressed != null) - { - BufferPool.put(compressed); - compressed = null; - } - } - } - - @Override - protected void reBufferStandard() - { - try - { - long position = current(); - assert position < metadata.dataLength; - - CompressionMetadata.Chunk chunk = metadata.chunkFor(position); - - if (compressed.capacity() < chunk.length) - { - BufferPool.put(compressed); - compressed = allocateBuffer(chunk.length, bufferType); - } - else - { - compressed.clear(); - } - - compressed.limit(chunk.length); - if (channel.read(compressed, chunk.offset) != chunk.length) - throw new CorruptBlockException(getPath(), chunk); - - compressed.flip(); - buffer.clear(); - - try - { - metadata.compressor().uncompress(compressed, buffer); - } - catch (IOException e) - { - throw new CorruptBlockException(getPath(), chunk); - } - finally - { - buffer.flip(); - } - - if (getCrcCheckChance() > ThreadLocalRandom.current().nextDouble()) - { - compressed.rewind(); - metadata.checksumType.update( checksum, (compressed)); - - if (checksum(chunk) != (int) checksum.getValue()) - throw new CorruptBlockException(getPath(), chunk); - - // reset checksum object back to the original (blank) state - checksum.reset(); - } - - // buffer offset is always aligned - bufferOffset = position & ~(buffer.capacity() - 1); - buffer.position((int) (position - bufferOffset)); - // the length() can be provided at construction time, to override the true (uncompressed) length of the file; - // this is permitted to occur within a compressed segment, so we truncate validBufferBytes if we cross the imposed length - if (bufferOffset + buffer.limit() > length()) - buffer.limit((int)(length() - bufferOffset)); - } - catch (CorruptBlockException e) - { - throw new CorruptSSTableException(e, getPath()); - } - catch (IOException e) - { - throw new FSReadError(e, getPath()); - } - } - - @Override - protected void reBufferMmap() - { - try - { - long position = current(); - assert position < metadata.dataLength; - - CompressionMetadata.Chunk chunk = metadata.chunkFor(position); - - MmappedRegions.Region region = regions.floor(chunk.offset); - long segmentOffset = region.bottom(); - int chunkOffset = Ints.checkedCast(chunk.offset - segmentOffset); - ByteBuffer compressedChunk = region.buffer.duplicate(); // TODO: change to slice(chunkOffset) when we upgrade LZ4-java - - compressedChunk.position(chunkOffset).limit(chunkOffset + chunk.length); - - buffer.clear(); - - try - { - metadata.compressor().uncompress(compressedChunk, buffer); - } - catch (IOException e) - { - throw new CorruptBlockException(getPath(), chunk); - } - finally - { - buffer.flip(); - } - - if (getCrcCheckChance() > ThreadLocalRandom.current().nextDouble()) - { - compressedChunk.position(chunkOffset).limit(chunkOffset + chunk.length); - - metadata.checksumType.update( checksum, compressedChunk); - - compressedChunk.limit(compressedChunk.capacity()); - if (compressedChunk.getInt() != (int) checksum.getValue()) - throw new CorruptBlockException(getPath(), chunk); - - // reset checksum object back to the original (blank) state - checksum.reset(); - } - - // buffer offset is always aligned - bufferOffset = position & ~(buffer.capacity() - 1); - buffer.position((int) (position - bufferOffset)); - // the length() can be provided at construction time, to override the true (uncompressed) length of the file; - // this is permitted to occur within a compressed segment, so we truncate validBufferBytes if we cross the imposed length - if (bufferOffset + buffer.limit() > length()) - buffer.limit((int)(length() - bufferOffset)); - } - catch (CorruptBlockException e) - { - throw new CorruptSSTableException(e, getPath()); - } - - } - - private int checksum(CompressionMetadata.Chunk chunk) throws IOException - { - long position = chunk.offset + chunk.length; - checksumBytes.clear(); - if (channel.read(checksumBytes, position) != checksumBytes.capacity()) - throw new CorruptBlockException(getPath(), chunk); - return checksumBytes.getInt(0); - } - - @Override - public long length() - { - return metadata.dataLength; - } - - @Override - public String toString() - { - return String.format("%s - chunk length %d, data length %d.", getPath(), metadata.chunkLength(), metadata.dataLength); - } - - public final static class Builder extends RandomAccessReader.Builder - { - private final CompressionMetadata metadata; - - public Builder(ICompressedFile file) - { - super(file.channel()); - this.metadata = applyMetadata(file.getMetadata()); - this.regions = file.regions(); - } - - public Builder(ChannelProxy channel, CompressionMetadata metadata) - { - super(channel); - this.metadata = applyMetadata(metadata); - } - - private CompressionMetadata applyMetadata(CompressionMetadata metadata) - { - this.overrideLength = metadata.compressedFileLength; - this.bufferSize = metadata.chunkLength(); - this.bufferType = metadata.compressor().preferredBufferType(); - - assert Integer.bitCount(this.bufferSize) == 1; //must be a power of two - - return metadata; - } - - @Override - protected ByteBuffer createBuffer() - { - buffer = allocateBuffer(bufferSize, bufferType); - buffer.limit(0); - return buffer; - } - - @Override - public RandomAccessReader build() - { - return new CompressedRandomAccessReader(this); - } - } -} diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index 3181a5504e..a77b673005 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -1637,7 +1637,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted lastEarlyOpenLength) + ChunkCache.instance.invalidatePosition(dfile, lastEarlyOpenLength); + lastEarlyOpenLength = dfile.dataLength(); + } + public SSTableReader openFinalEarly() { // we must ensure the data is completely flushed to disk @@ -296,6 +308,7 @@ public class BigTableWriter extends SSTableWriter IndexSummary indexSummary = iwriter.summary.build(this.metadata.partitioner); SegmentedFile ifile = iwriter.builder.buildIndex(desc, indexSummary); SegmentedFile dfile = dbuilder.buildData(desc, stats); + invalidateCacheAtBoundary(dfile); SSTableReader sstable = SSTableReader.internalOpen(desc, components, this.metadata, diff --git a/src/java/org/apache/cassandra/io/util/AbstractReaderFileProxy.java b/src/java/org/apache/cassandra/io/util/AbstractReaderFileProxy.java new file mode 100644 index 0000000000..5dc0d37a4a --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/AbstractReaderFileProxy.java @@ -0,0 +1,61 @@ +/* + * 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 abstract class AbstractReaderFileProxy implements ReaderFileProxy +{ + protected final ChannelProxy channel; + protected final long fileLength; + + public AbstractReaderFileProxy(ChannelProxy channel, long fileLength) + { + this.channel = channel; + this.fileLength = fileLength >= 0 ? fileLength : channel.size(); + } + + @Override + public ChannelProxy channel() + { + return channel; + } + + @Override + public long fileLength() + { + return fileLength; + } + + @Override + public String toString() + { + return getClass().getSimpleName() + "(filePath='" + channel + "')"; + } + + @Override + public void close() + { + // nothing in base class + } + + @Override + public double getCrcCheckChance() + { + return 0; // Only valid for compressed files. + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/util/BufferManagingRebufferer.java b/src/java/org/apache/cassandra/io/util/BufferManagingRebufferer.java new file mode 100644 index 0000000000..0e67364e8c --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/BufferManagingRebufferer.java @@ -0,0 +1,127 @@ +package org.apache.cassandra.io.util; + +import java.nio.ByteBuffer; + +import org.apache.cassandra.utils.memory.BufferPool; + +/** + * Buffer manager used for reading from a ChunkReader when cache is not in use. Instances of this class are + * reader-specific and thus do not need to be thread-safe since the reader itself isn't. + * + * The instances reuse themselves as the BufferHolder to avoid having to return a new object for each rebuffer call. + */ +public abstract class BufferManagingRebufferer implements Rebufferer, Rebufferer.BufferHolder +{ + protected final ChunkReader source; + protected final ByteBuffer buffer; + protected long offset = 0; + + public static BufferManagingRebufferer on(ChunkReader wrapped) + { + return wrapped.alignmentRequired() + ? new Aligned(wrapped) + : new Unaligned(wrapped); + } + + abstract long alignedPosition(long position); + + public BufferManagingRebufferer(ChunkReader wrapped) + { + this.source = wrapped; + buffer = RandomAccessReader.allocateBuffer(wrapped.chunkSize(), wrapped.preferredBufferType()); + buffer.limit(0); + } + + @Override + public void closeReader() + { + BufferPool.put(buffer); + offset = -1; + } + + @Override + public void close() + { + assert offset == -1; // reader must be closed at this point. + source.close(); + } + + @Override + public ChannelProxy channel() + { + return source.channel(); + } + + @Override + public long fileLength() + { + return source.fileLength(); + } + + @Override + public BufferHolder rebuffer(long position) + { + offset = alignedPosition(position); + source.readChunk(offset, buffer); + return this; + } + + @Override + public double getCrcCheckChance() + { + return source.getCrcCheckChance(); + } + + @Override + public String toString() + { + return "BufferManagingRebufferer." + getClass().getSimpleName() + ":" + source.toString(); + } + + // BufferHolder methods + + public ByteBuffer buffer() + { + return buffer; + } + + public long offset() + { + return offset; + } + + @Override + public void release() + { + // nothing to do, we don't delete buffers before we're closed. + } + + public static class Unaligned extends BufferManagingRebufferer + { + public Unaligned(ChunkReader wrapped) + { + super(wrapped); + } + + @Override + long alignedPosition(long position) + { + return position; + } + } + + public static class Aligned extends BufferManagingRebufferer + { + public Aligned(ChunkReader wrapped) + { + super(wrapped); + assert Integer.bitCount(wrapped.chunkSize()) == 1; + } + + @Override + long alignedPosition(long position) + { + return position & -buffer.capacity(); + } + } +} diff --git a/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java index 090c5bd198..a46ec14796 100644 --- a/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java @@ -17,11 +17,19 @@ */ package org.apache.cassandra.io.util; +import org.apache.cassandra.cache.ChunkCache; +import org.apache.cassandra.io.compress.BufferType; + public class BufferedSegmentedFile extends SegmentedFile { public BufferedSegmentedFile(ChannelProxy channel, int bufferSize, long length) { - super(new Cleanup(channel), channel, bufferSize, length); + this(channel, createRebufferer(channel, length, bufferSize), length); + } + + private BufferedSegmentedFile(ChannelProxy channel, RebuffererFactory rebufferer, long length) + { + super(new Cleanup(channel, rebufferer), channel, rebufferer, length); } private BufferedSegmentedFile(BufferedSegmentedFile copy) @@ -29,6 +37,11 @@ public class BufferedSegmentedFile extends SegmentedFile super(copy); } + private static RebuffererFactory createRebufferer(ChannelProxy channel, long length, int bufferSize) + { + return ChunkCache.maybeWrap(new SimpleChunkReader(channel, length, BufferType.OFF_HEAP, bufferSize)); + } + public static class Builder extends SegmentedFile.Builder { public SegmentedFile complete(ChannelProxy channel, int bufferSize, long overrideLength) diff --git a/src/java/org/apache/cassandra/io/util/ChannelProxy.java b/src/java/org/apache/cassandra/io/util/ChannelProxy.java index f8661606b9..361b0d32f8 100644 --- a/src/java/org/apache/cassandra/io/util/ChannelProxy.java +++ b/src/java/org/apache/cassandra/io/util/ChannelProxy.java @@ -125,6 +125,7 @@ public final class ChannelProxy extends SharedCloseableImpl { try { + // FIXME: consider wrapping in a while loop return channel.read(buffer, position); } catch (IOException e) diff --git a/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java b/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java index 30f1e0c1fa..25ef6158ef 100644 --- a/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java @@ -19,13 +19,13 @@ package org.apache.cassandra.io.util; import java.io.File; import java.io.IOException; -import java.util.zip.CRC32; import org.apache.cassandra.io.compress.BufferType; +import org.apache.cassandra.io.util.DataIntegrityMetadata.ChecksumValidator; import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.Throwables; +import org.apache.cassandra.utils.ChecksumType; -public class ChecksummedRandomAccessReader extends RandomAccessReader +public class ChecksummedRandomAccessReader { @SuppressWarnings("serial") public static class CorruptFileException extends RuntimeException @@ -39,67 +39,56 @@ public class ChecksummedRandomAccessReader extends RandomAccessReader } } - private final DataIntegrityMetadata.ChecksumValidator validator; - - private ChecksummedRandomAccessReader(Builder builder) + static class ChecksummedRebufferer extends BufferManagingRebufferer { - super(builder); - this.validator = builder.validator; - } + private final DataIntegrityMetadata.ChecksumValidator validator; - @SuppressWarnings("resource") - @Override - protected void reBufferStandard() - { - long desiredPosition = current(); - // align with buffer size, as checksums were computed in chunks of buffer size each. - bufferOffset = (desiredPosition / buffer.capacity()) * buffer.capacity(); - - buffer.clear(); - - long position = bufferOffset; - while (buffer.hasRemaining()) + public ChecksummedRebufferer(ChannelProxy channel, ChecksumValidator validator) { - int n = channel.read(buffer, position); - if (n < 0) - break; - position += n; + super(new SimpleChunkReader(channel, channel.size(), BufferType.ON_HEAP, validator.chunkSize)); + this.validator = validator; } - buffer.flip(); + @Override + public BufferHolder rebuffer(long desiredPosition) + { + if (desiredPosition != offset + buffer.position()) + validator.seek(desiredPosition); - try - { - validator.validate(ByteBufferUtil.getArray(buffer), 0, buffer.remaining()); - } - catch (IOException e) - { - throw new CorruptFileException(e, channel.filePath()); + // align with buffer size, as checksums were computed in chunks of buffer size each. + offset = alignedPosition(desiredPosition); + source.readChunk(offset, buffer); + + try + { + validator.validate(ByteBufferUtil.getArray(buffer), 0, buffer.remaining()); + } + catch (IOException e) + { + throw new CorruptFileException(e, channel().filePath()); + } + + return this; } - buffer.position((int) (desiredPosition - bufferOffset)); - } + @Override + public void close() + { + try + { + source.close(); + } + finally + { + validator.close(); + } + } - @Override - protected void reBufferMmap() - { - throw new AssertionError("Unsupported operation"); - } - - @Override - public void seek(long newPosition) - { - validator.seek(newPosition); - super.seek(newPosition); - } - - @Override - public void close() - { - Throwables.perform(channel.filePath(), Throwables.FileOpType.READ, - super::close, - validator::close, - channel::close); + @Override + long alignedPosition(long desiredPosition) + { + return (desiredPosition / buffer.capacity()) * buffer.capacity(); + } } public static final class Builder extends RandomAccessReader.Builder @@ -110,18 +99,22 @@ public class ChecksummedRandomAccessReader extends RandomAccessReader public Builder(File file, File crcFile) throws IOException { super(new ChannelProxy(file)); - this.validator = new DataIntegrityMetadata.ChecksumValidator(new CRC32(), + this.validator = new DataIntegrityMetadata.ChecksumValidator(ChecksumType.CRC32, RandomAccessReader.open(crcFile), file.getPath()); + } - super.bufferSize(validator.chunkSize) - .bufferType(BufferType.ON_HEAP); + @Override + protected Rebufferer createRebufferer() + { + return new ChecksummedRebufferer(channel, validator); } @Override public RandomAccessReader build() { - return new ChecksummedRandomAccessReader(this); + // Always own and close the channel. + return buildWithChannel(); } } } diff --git a/src/java/org/apache/cassandra/io/util/ChunkReader.java b/src/java/org/apache/cassandra/io/util/ChunkReader.java new file mode 100644 index 0000000000..a04299a74a --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/ChunkReader.java @@ -0,0 +1,56 @@ +/* + * 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.ByteBuffer; + +import org.apache.cassandra.io.compress.BufferType; + +/** + * RandomFileReader component that reads data from a file into a provided buffer and may have requirements over the + * size and alignment of reads. + * A caching or buffer-managing rebufferer will reference one of these to do the actual reading. + * Note: Implementations of this interface must be thread-safe! + */ +public interface ChunkReader extends RebuffererFactory +{ + /** + * Read the chunk at the given position, attempting to fill the capacity of the given buffer. + * The filled buffer must be positioned at 0, with limit set at the size of the available data. + * The source may have requirements for the positioning and/or size of the buffer (e.g. chunk-aligned and + * chunk-sized). These must be satisfied by the caller. + */ + void readChunk(long position, ByteBuffer buffer); + + /** + * Buffer size required for this rebufferer. Must be power of 2 if alignment is required. + */ + int chunkSize(); + + /** + * If true, positions passed to this rebufferer must be aligned to chunkSize. + */ + boolean alignmentRequired(); + + /** + * Specifies type of buffer the caller should attempt to give. + * This is not guaranteed to be fulfilled. + */ + BufferType preferredBufferType(); +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java index 16f791a43d..7365d40c3e 100644 --- a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java @@ -17,49 +17,61 @@ */ package org.apache.cassandra.io.util; -import com.google.common.util.concurrent.RateLimiter; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.concurrent.ThreadLocalRandom; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.primitives.Ints; +import org.apache.cassandra.cache.ChunkCache; import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.Config.DiskAccessMode; +import org.apache.cassandra.io.compress.*; +import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.io.compress.CompressedRandomAccessReader; -import org.apache.cassandra.io.compress.CompressedSequentialWriter; -import org.apache.cassandra.io.compress.CompressionMetadata; -import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.concurrent.Ref; public class CompressedSegmentedFile extends SegmentedFile implements ICompressedFile { - private static final Logger logger = LoggerFactory.getLogger(CompressedSegmentedFile.class); - private static final boolean useMmap = DatabaseDescriptor.getDiskAccessMode() == Config.DiskAccessMode.mmap; - public final CompressionMetadata metadata; - private final MmappedRegions regions; - public CompressedSegmentedFile(ChannelProxy channel, int bufferSize, CompressionMetadata metadata) + public CompressedSegmentedFile(ChannelProxy channel, CompressionMetadata metadata, Config.DiskAccessMode mode) { this(channel, - bufferSize, metadata, - useMmap + mode == DiskAccessMode.mmap ? MmappedRegions.map(channel, metadata) : null); } - public CompressedSegmentedFile(ChannelProxy channel, int bufferSize, CompressionMetadata metadata, MmappedRegions regions) + public CompressedSegmentedFile(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions) { - super(new Cleanup(channel, metadata, regions), channel, bufferSize, metadata.dataLength, metadata.compressedFileLength); + this(channel, metadata, regions, createRebufferer(channel, metadata, regions)); + } + + private static RebuffererFactory createRebufferer(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions) + { + return ChunkCache.maybeWrap(chunkReader(channel, metadata, regions)); + } + + public static ChunkReader chunkReader(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions) + { + return regions != null + ? new Mmap(channel, metadata, regions) + : new Standard(channel, metadata); + } + + public CompressedSegmentedFile(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions, RebuffererFactory rebufferer) + { + super(new Cleanup(channel, metadata, regions, rebufferer), channel, rebufferer, metadata.compressedFileLength); this.metadata = metadata; - this.regions = regions; } private CompressedSegmentedFile(CompressedSegmentedFile copy) { super(copy); this.metadata = copy.metadata; - this.regions = copy.regions; } public ChannelProxy channel() @@ -67,33 +79,21 @@ public class CompressedSegmentedFile extends SegmentedFile implements ICompresse return channel; } - public MmappedRegions regions() - { - return regions; - } - private static final class Cleanup extends SegmentedFile.Cleanup { final CompressionMetadata metadata; - private final MmappedRegions regions; - protected Cleanup(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions) + protected Cleanup(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions, ReaderFileProxy rebufferer) { - super(channel); + super(channel, rebufferer); this.metadata = metadata; - this.regions = regions; } public void tidy() { - Throwable err = regions == null ? null : regions.close(null); - if (err != null) + if (ChunkCache.instance != null) { - JVMStabilityInspector.inspectThrowable(err); - - // This is not supposed to happen - logger.error("Error while closing mmapped regions", err); + ChunkCache.instance.invalidateFile(name()); } - metadata.close(); super.tidy(); @@ -114,9 +114,12 @@ public class CompressedSegmentedFile extends SegmentedFile implements ICompresse public static class Builder extends SegmentedFile.Builder { final CompressedSequentialWriter writer; + final Config.DiskAccessMode mode; + public Builder(CompressedSequentialWriter writer) { this.writer = writer; + this.mode = DatabaseDescriptor.getDiskAccessMode(); } protected CompressionMetadata metadata(String path, long overrideLength) @@ -129,7 +132,7 @@ public class CompressedSegmentedFile extends SegmentedFile implements ICompresse public SegmentedFile complete(ChannelProxy channel, int bufferSize, long overrideLength) { - return new CompressedSegmentedFile(channel, bufferSize, metadata(channel.filePath(), overrideLength)); + return new CompressedSegmentedFile(channel, metadata(channel.filePath(), overrideLength), mode); } } @@ -140,18 +143,216 @@ public class CompressedSegmentedFile extends SegmentedFile implements ICompresse super.dropPageCache(metadata.chunkFor(before).offset); } - public RandomAccessReader createReader() - { - return new CompressedRandomAccessReader.Builder(this).build(); - } - - public RandomAccessReader createReader(RateLimiter limiter) - { - return new CompressedRandomAccessReader.Builder(this).limiter(limiter).build(); - } - public CompressionMetadata getMetadata() { return metadata; } + + public long dataLength() + { + return metadata.dataLength; + } + + @VisibleForTesting + public abstract static class CompressedChunkReader extends AbstractReaderFileProxy implements ChunkReader + { + final CompressionMetadata metadata; + + public CompressedChunkReader(ChannelProxy channel, CompressionMetadata metadata) + { + super(channel, metadata.dataLength); + this.metadata = metadata; + assert Integer.bitCount(metadata.chunkLength()) == 1; //must be a power of two + } + + @VisibleForTesting + public double getCrcCheckChance() + { + return metadata.parameters.getCrcCheckChance(); + } + + @Override + public String toString() + { + return String.format("CompressedChunkReader.%s(%s - %s, chunk length %d, data length %d)", + getClass().getSimpleName(), + channel.filePath(), + metadata.compressor().getClass().getSimpleName(), + metadata.chunkLength(), + metadata.dataLength); + } + + @Override + public int chunkSize() + { + return metadata.chunkLength(); + } + + @Override + public boolean alignmentRequired() + { + return true; + } + + @Override + public BufferType preferredBufferType() + { + return metadata.compressor().preferredBufferType(); + } + + @Override + public Rebufferer instantiateRebufferer() + { + return BufferManagingRebufferer.on(this); + } + } + + static class Standard extends CompressedChunkReader + { + // we read the raw compressed bytes into this buffer, then uncompressed them into the provided one. + private final ThreadLocal compressedHolder; + + public Standard(ChannelProxy channel, CompressionMetadata metadata) + { + super(channel, metadata); + compressedHolder = ThreadLocal.withInitial(this::allocateBuffer); + } + + public ByteBuffer allocateBuffer() + { + return allocateBuffer(metadata.compressor().initialCompressedBufferLength(metadata.chunkLength())); + } + + public ByteBuffer allocateBuffer(int size) + { + return metadata.compressor().preferredBufferType().allocate(size); + } + + @Override + public void readChunk(long position, ByteBuffer uncompressed) + { + try + { + // accesses must always be aligned + assert (position & -uncompressed.capacity()) == position; + assert position <= fileLength; + + CompressionMetadata.Chunk chunk = metadata.chunkFor(position); + ByteBuffer compressed = compressedHolder.get(); + + if (compressed.capacity() < chunk.length) + { + compressed = allocateBuffer(chunk.length); + compressedHolder.set(compressed); + } + else + { + compressed.clear(); + } + + compressed.limit(chunk.length); + if (channel.read(compressed, chunk.offset) != chunk.length) + throw new CorruptBlockException(channel.filePath(), chunk); + + compressed.flip(); + uncompressed.clear(); + + try + { + metadata.compressor().uncompress(compressed, uncompressed); + } + catch (IOException e) + { + throw new CorruptBlockException(channel.filePath(), chunk); + } + finally + { + uncompressed.flip(); + } + + if (getCrcCheckChance() > ThreadLocalRandom.current().nextDouble()) + { + compressed.rewind(); + int checksum = (int) metadata.checksumType.of(compressed); + + compressed.clear().limit(Integer.BYTES); + if (channel.read(compressed, chunk.offset + chunk.length) != Integer.BYTES + || compressed.getInt(0) != checksum) + throw new CorruptBlockException(channel.filePath(), chunk); + } + } + catch (CorruptBlockException e) + { + throw new CorruptSSTableException(e, channel.filePath()); + } + } + } + + static class Mmap extends CompressedChunkReader + { + protected final MmappedRegions regions; + + public Mmap(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions) + { + super(channel, metadata); + this.regions = regions; + } + + @Override + public void readChunk(long position, ByteBuffer uncompressed) + { + try + { + // accesses must always be aligned + assert (position & -uncompressed.capacity()) == position; + assert position <= fileLength; + + CompressionMetadata.Chunk chunk = metadata.chunkFor(position); + + MmappedRegions.Region region = regions.floor(chunk.offset); + long segmentOffset = region.offset(); + int chunkOffset = Ints.checkedCast(chunk.offset - segmentOffset); + ByteBuffer compressedChunk = region.buffer(); + + compressedChunk.position(chunkOffset).limit(chunkOffset + chunk.length); + + uncompressed.clear(); + + try + { + metadata.compressor().uncompress(compressedChunk, uncompressed); + } + catch (IOException e) + { + throw new CorruptBlockException(channel.filePath(), chunk); + } + finally + { + uncompressed.flip(); + } + + if (getCrcCheckChance() > ThreadLocalRandom.current().nextDouble()) + { + compressedChunk.position(chunkOffset).limit(chunkOffset + chunk.length); + + int checksum = (int) metadata.checksumType.of(compressedChunk); + + compressedChunk.limit(compressedChunk.capacity()); + if (compressedChunk.getInt() != checksum) + throw new CorruptBlockException(channel.filePath(), chunk); + } + } + catch (CorruptBlockException e) + { + throw new CorruptSSTableException(e, channel.filePath()); + } + + } + + public void close() + { + regions.closeQuietly(); + super.close(); + } + } } diff --git a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java index eb84a891fc..0c48d13397 100644 --- a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java +++ b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java @@ -34,6 +34,7 @@ import com.google.common.base.Charsets; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.utils.ChecksumType; import org.apache.cassandra.utils.Throwables; public class DataIntegrityMetadata @@ -45,21 +46,21 @@ public class DataIntegrityMetadata public static class ChecksumValidator implements Closeable { - private final Checksum checksum; + private final ChecksumType checksumType; private final RandomAccessReader reader; public final int chunkSize; private final String dataFilename; public ChecksumValidator(Descriptor descriptor) throws IOException { - this(descriptor.version.uncompressedChecksumType().newInstance(), + this(descriptor.version.uncompressedChecksumType(), RandomAccessReader.open(new File(descriptor.filenameFor(Component.CRC))), descriptor.filenameFor(Component.DATA)); } - public ChecksumValidator(Checksum checksum, RandomAccessReader reader, String dataFilename) throws IOException + public ChecksumValidator(ChecksumType checksumType, RandomAccessReader reader, String dataFilename) throws IOException { - this.checksum = checksum; + this.checksumType = checksumType; this.reader = reader; this.dataFilename = dataFilename; chunkSize = reader.readInt(); @@ -79,9 +80,7 @@ public class DataIntegrityMetadata public void validate(byte[] bytes, int start, int end) throws IOException { - checksum.update(bytes, start, end); - int current = (int) checksum.getValue(); - checksum.reset(); + int current = (int) checksumType.of(bytes, start, end); int actual = reader.readInt(); if (current != actual) throw new IOException("Corrupted File : " + dataFilename); diff --git a/src/java/org/apache/cassandra/io/util/ICompressedFile.java b/src/java/org/apache/cassandra/io/util/ICompressedFile.java index 43cef8c4e3..e69487c801 100644 --- a/src/java/org/apache/cassandra/io/util/ICompressedFile.java +++ b/src/java/org/apache/cassandra/io/util/ICompressedFile.java @@ -23,6 +23,4 @@ public interface ICompressedFile { ChannelProxy channel(); CompressionMetadata getMetadata(); - MmappedRegions regions(); - } diff --git a/src/java/org/apache/cassandra/io/util/LimitingRebufferer.java b/src/java/org/apache/cassandra/io/util/LimitingRebufferer.java new file mode 100644 index 0000000000..e69da70754 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/LimitingRebufferer.java @@ -0,0 +1,106 @@ +package org.apache.cassandra.io.util; + +import java.nio.ByteBuffer; + +import com.google.common.primitives.Ints; +import com.google.common.util.concurrent.RateLimiter; + +/** + * Rebufferer wrapper that applies rate limiting. + * + * Instantiated once per RandomAccessReader, thread-unsafe. + * The instances reuse themselves as the BufferHolder to avoid having to return a new object for each rebuffer call. + */ +public class LimitingRebufferer implements Rebufferer, Rebufferer.BufferHolder +{ + final private Rebufferer wrapped; + final private RateLimiter limiter; + final private int limitQuant; + + private BufferHolder bufferHolder; + private ByteBuffer buffer; + private long offset; + + public LimitingRebufferer(Rebufferer wrapped, RateLimiter limiter, int limitQuant) + { + this.wrapped = wrapped; + this.limiter = limiter; + this.limitQuant = limitQuant; + } + + @Override + public BufferHolder rebuffer(long position) + { + bufferHolder = wrapped.rebuffer(position); + buffer = bufferHolder.buffer(); + offset = bufferHolder.offset(); + int posInBuffer = Ints.checkedCast(position - offset); + int remaining = buffer.limit() - posInBuffer; + if (remaining == 0) + return this; + + if (remaining > limitQuant) + { + buffer.limit(posInBuffer + limitQuant); // certainly below current limit + remaining = limitQuant; + } + limiter.acquire(remaining); + return this; + } + + @Override + public ChannelProxy channel() + { + return wrapped.channel(); + } + + @Override + public long fileLength() + { + return wrapped.fileLength(); + } + + @Override + public double getCrcCheckChance() + { + return wrapped.getCrcCheckChance(); + } + + @Override + public void close() + { + wrapped.close(); + } + + @Override + public void closeReader() + { + wrapped.closeReader(); + } + + @Override + public String toString() + { + return "LimitingRebufferer[" + limiter.toString() + "]:" + wrapped.toString(); + } + + // BufferHolder methods + + @Override + public ByteBuffer buffer() + { + return buffer; + } + + @Override + public long offset() + { + return offset; + } + + @Override + public void release() + { + bufferHolder.release(); + } +} diff --git a/src/java/org/apache/cassandra/io/util/MmapRebufferer.java b/src/java/org/apache/cassandra/io/util/MmapRebufferer.java new file mode 100644 index 0000000000..6c39cb1dbe --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/MmapRebufferer.java @@ -0,0 +1,49 @@ +package org.apache.cassandra.io.util; + +/** + * Rebufferer for memory-mapped files. Thread-safe and shared among reader instances. + * This is simply a thin wrapper around MmappedRegions as the buffers there can be used directly after duplication. + */ +class MmapRebufferer extends AbstractReaderFileProxy implements Rebufferer, RebuffererFactory +{ + protected final MmappedRegions regions; + + public MmapRebufferer(ChannelProxy channel, long fileLength, MmappedRegions regions) + { + super(channel, fileLength); + this.regions = regions; + } + + @Override + public BufferHolder rebuffer(long position) + { + return regions.floor(position); + } + + @Override + public Rebufferer instantiateRebufferer() + { + return this; + } + + @Override + public void close() + { + regions.closeQuietly(); + } + + @Override + public void closeReader() + { + // Instance is shared among readers. Nothing to release. + } + + @Override + public String toString() + { + return String.format("%s(%s - data length %d)", + getClass().getSimpleName(), + channel.filePath(), + fileLength()); + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/util/MmappedRegions.java b/src/java/org/apache/cassandra/io/util/MmappedRegions.java index 8f6cd92eb5..f269b84c5f 100644 --- a/src/java/org/apache/cassandra/io/util/MmappedRegions.java +++ b/src/java/org/apache/cassandra/io/util/MmappedRegions.java @@ -22,8 +22,11 @@ import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.util.Arrays; +import org.slf4j.LoggerFactory; + import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.io.compress.CompressionMetadata; +import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.RefCounted; import org.apache.cassandra.utils.concurrent.SharedCloseableImpl; @@ -190,8 +193,20 @@ public class MmappedRegions extends SharedCloseableImpl assert !isCleanedUp() : "Attempted to use closed region"; return state.floor(position); } + + public void closeQuietly() + { + Throwable err = close(null); + if (err != null) + { + JVMStabilityInspector.inspectThrowable(err); - public static final class Region + // This is not supposed to happen + LoggerFactory.getLogger(getClass()).error("Error while closing mmapped regions", err); + } + } + + public static final class Region implements Rebufferer.BufferHolder { public final long offset; public final ByteBuffer buffer; @@ -202,15 +217,25 @@ public class MmappedRegions extends SharedCloseableImpl this.buffer = buffer; } - public long bottom() + public ByteBuffer buffer() + { + return buffer.duplicate(); + } + + public long offset() { return offset; } - public long top() + public long end() { return offset + buffer.capacity(); } + + public void release() + { + // only released after no readers are present + } } private static final class State @@ -260,7 +285,7 @@ public class MmappedRegions extends SharedCloseableImpl private Region floor(long position) { - assert 0 <= position && position < length : String.format("%d >= %d", position, length); + assert 0 <= position && position <= length : String.format("%d > %d", position, length); int idx = Arrays.binarySearch(offsets, 0, last +1, position); assert idx != -1 : String.format("Bad position %d for regions %s, last %d in %s", position, Arrays.toString(offsets), last, channel); diff --git a/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java index 5f56ff64a1..d514bf8022 100644 --- a/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java @@ -19,30 +19,29 @@ package org.apache.cassandra.io.util; import java.io.*; -import com.google.common.util.concurrent.RateLimiter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.sstable.format.Version; -import org.apache.cassandra.utils.JVMStabilityInspector; public class MmappedSegmentedFile extends SegmentedFile { private static final Logger logger = LoggerFactory.getLogger(MmappedSegmentedFile.class); - private final MmappedRegions regions; - - public MmappedSegmentedFile(ChannelProxy channel, int bufferSize, long length, MmappedRegions regions) + public MmappedSegmentedFile(ChannelProxy channel, long length, MmappedRegions regions) { - super(new Cleanup(channel, regions), channel, bufferSize, length); - this.regions = regions; + this(channel, new MmapRebufferer(channel, length, regions), length); + } + + public MmappedSegmentedFile(ChannelProxy channel, RebuffererFactory rebufferer, long length) + { + super(new Cleanup(channel, rebufferer), channel, rebufferer, length); } private MmappedSegmentedFile(MmappedSegmentedFile copy) { super(copy); - this.regions = copy.regions; } public MmappedSegmentedFile sharedCopy() @@ -50,49 +49,6 @@ public class MmappedSegmentedFile extends SegmentedFile return new MmappedSegmentedFile(this); } - public RandomAccessReader createReader() - { - return new RandomAccessReader.Builder(channel) - .overrideLength(length) - .regions(regions) - .build(); - } - - public RandomAccessReader createReader(RateLimiter limiter) - { - return new RandomAccessReader.Builder(channel) - .overrideLength(length) - .bufferSize(bufferSize) - .regions(regions) - .limiter(limiter) - .build(); - } - - private static final class Cleanup extends SegmentedFile.Cleanup - { - private final MmappedRegions regions; - - Cleanup(ChannelProxy channel, MmappedRegions regions) - { - super(channel); - this.regions = regions; - } - - public void tidy() - { - Throwable err = regions.close(null); - if (err != null) - { - JVMStabilityInspector.inspectThrowable(err); - - // This is not supposed to happen - logger.error("Error while closing mmapped regions", err); - } - - super.tidy(); - } - } - /** * Overrides the default behaviour to create segments of a maximum size. */ @@ -110,7 +66,7 @@ public class MmappedSegmentedFile extends SegmentedFile long length = overrideLength > 0 ? overrideLength : channel.size(); updateRegions(channel, length); - return new MmappedSegmentedFile(channel, bufferSize, length, regions.sharedCopy()); + return new MmappedSegmentedFile(channel, length, regions.sharedCopy()); } private void updateRegions(ChannelProxy channel, long length) diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index e6383cf80c..725b3674b7 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -21,11 +21,13 @@ import java.io.*; import java.nio.ByteBuffer; import java.nio.ByteOrder; +import com.google.common.annotations.VisibleForTesting; import com.google.common.primitives.Ints; import com.google.common.util.concurrent.RateLimiter; -import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.io.compress.BufferType; +import org.apache.cassandra.io.compress.CompressionMetadata; +import org.apache.cassandra.io.util.Rebufferer.BufferHolder; import org.apache.cassandra.utils.memory.BufferPool; public class RandomAccessReader extends RebufferingInputStream implements FileDataInput @@ -41,62 +43,24 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa // and because our BufferPool currently has a maximum allocation size of this. public static final int MAX_BUFFER_SIZE = 1 << 16; // 64k - // 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; - - // optional memory mapped regions for the channel - protected final MmappedRegions regions; - - // An optional limiter that will throttle the amount of data we read - protected final RateLimiter limiter; - - // the file length, 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, - // required when opening sstables early not to read past the mark - private final long fileLength; - - // the buffer size for buffered readers - protected final int bufferSize; - - // the buffer type for buffered readers - protected final BufferType bufferType; - - // offset from the beginning of the file - protected long bufferOffset; - // offset of the last file mark protected long markedPointer; - protected RandomAccessReader(Builder builder) - { - super(builder.createBuffer()); + @VisibleForTesting + final Rebufferer rebufferer; + BufferHolder bufferHolder = Rebufferer.EMPTY; - this.channel = builder.channel; - this.regions = builder.regions; - this.limiter = builder.limiter; - this.fileLength = builder.overrideLength <= 0 ? builder.channel.size() : builder.overrideLength; - this.bufferSize = builder.bufferSize; - this.bufferType = builder.bufferType; - this.buffer = builder.buffer; + protected RandomAccessReader(Rebufferer rebufferer) + { + super(Rebufferer.EMPTY.buffer()); + this.rebufferer = rebufferer; } - protected static ByteBuffer allocateBuffer(int size, BufferType bufferType) + public static ByteBuffer allocateBuffer(int size, BufferType bufferType) { return BufferPool.get(size, bufferType).order(ByteOrder.BIG_ENDIAN); } - protected void releaseBuffer() - { - if (buffer != null) - { - if (regions == null) - BufferPool.put(buffer); - buffer = null; - } - } - /** * Read data from file starting from current currentOffset to populate buffer. */ @@ -105,80 +69,40 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa if (isEOF()) return; - if (regions == null) - reBufferStandard(); - else - reBufferMmap(); + reBufferAt(current()); + } - if (limiter != null) - limiter.acquire(buffer.remaining()); + public void reBufferAt(long position) + { + bufferHolder.release(); + bufferHolder = rebufferer.rebuffer(position); + buffer = bufferHolder.buffer(); + buffer.position(Ints.checkedCast(position - bufferHolder.offset())); assert buffer.order() == ByteOrder.BIG_ENDIAN : "Buffer must have BIG ENDIAN byte ordering"; } - protected void reBufferStandard() - { - bufferOffset += buffer.position(); - assert bufferOffset < fileLength; - - buffer.clear(); - long position = bufferOffset; - long limit = bufferOffset; - - long pageAligedPos = position & ~4095; - // Because the buffer capacity is a multiple of the page size, we read less - // the first time and then we should read at page boundaries only, - // unless the user seeks elsewhere - long upperLimit = Math.min(fileLength, pageAligedPos + buffer.capacity()); - buffer.limit((int)(upperLimit - position)); - while (buffer.hasRemaining() && limit < upperLimit) - { - int n = channel.read(buffer, position); - if (n < 0) - throw new FSReadError(new IOException("Unexpected end of file"), channel.filePath()); - - position += n; - limit = bufferOffset + buffer.position(); - } - - buffer.flip(); - } - - protected void reBufferMmap() - { - long position = bufferOffset + buffer.position(); - assert position < fileLength; - - MmappedRegions.Region region = regions.floor(position); - bufferOffset = region.bottom(); - buffer = region.buffer.duplicate(); - buffer.position(Ints.checkedCast(position - bufferOffset)); - - if (limiter != null && bufferSize < buffer.remaining()) - { // ensure accurate throttling - buffer.limit(buffer.position() + bufferSize); - } - } - @Override public long getFilePointer() { + if (buffer == null) // closed already + return rebufferer.fileLength(); return current(); } protected long current() { - return bufferOffset + (buffer == null ? 0 : buffer.position()); + return bufferHolder.offset() + buffer.position(); } public String getPath() { - return channel.filePath(); + return getChannel().filePath(); } public ChannelProxy getChannel() { - return channel; + return rebufferer.channel(); } @Override @@ -242,12 +166,14 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa @Override public void close() { - //make idempotent + // close needs to be idempotent. if (buffer == null) return; - bufferOffset += buffer.position(); - releaseBuffer(); + bufferHolder.release(); + rebufferer.closeReader(); + buffer = null; + bufferHolder = null; //For performance reasons we don't keep a reference to the file //channel so we don't close it @@ -256,7 +182,7 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa @Override public String toString() { - return getClass().getSimpleName() + "(filePath='" + channel + "')"; + return getClass().getSimpleName() + ':' + rebufferer.toString(); } /** @@ -281,26 +207,17 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa 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()) - throw new IllegalArgumentException(String.format("Unable to seek to position %d in %s (%d bytes) in read-only mode", - newPosition, getPath(), length())); - buffer.limit(0); - bufferOffset = newPosition; - return; - } - + long bufferOffset = bufferHolder.offset(); if (newPosition >= bufferOffset && newPosition < bufferOffset + buffer.limit()) { buffer.position((int) (newPosition - bufferOffset)); return; } - // Set current location to newPosition and clear buffer so reBuffer calculates from newPosition - bufferOffset = newPosition; - buffer.clear(); - reBuffer(); - assert current() == newPosition; + + if (newPosition > length()) + throw new IllegalArgumentException(String.format("Unable to seek to position %d in %s (%d bytes) in read-only mode", + newPosition, getPath(), length())); + reBufferAt(newPosition); } /** @@ -353,7 +270,7 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa public long length() { - return fileLength; + return rebufferer.fileLength(); } public long getPosition() @@ -361,17 +278,38 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa return current(); } + public double getCrcCheckChance() + { + return rebufferer.getCrcCheckChance(); + } + + protected static Rebufferer instantiateRebufferer(RebuffererFactory fileRebufferer, RateLimiter limiter) + { + Rebufferer rebufferer = fileRebufferer.instantiateRebufferer(); + + if (limiter != null) + rebufferer = new LimitingRebufferer(rebufferer, limiter, MAX_BUFFER_SIZE); + + return rebufferer; + } + + public static RandomAccessReader build(SegmentedFile file, RateLimiter limiter) + { + return new RandomAccessReader(instantiateRebufferer(file.rebuffererFactory(), limiter)); + } + + public static Builder builder(ChannelProxy channel) + { + return new Builder(channel); + } + public static class Builder { // The NIO file channel or an empty channel public final ChannelProxy channel; - // We override the file length when we open sstables early, so that we do not - // read past the early mark - public long overrideLength; - // The size of the buffer for buffered readers - public int bufferSize; + protected int bufferSize; // The type of the buffer for buffered readers public BufferType bufferType; @@ -379,20 +317,20 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa // The buffer public ByteBuffer buffer; + // An optional limiter that will throttle the amount of data we read + public RateLimiter limiter; + // The mmap segments for mmap readers public MmappedRegions regions; - // An optional limiter that will throttle the amount of data we read - public RateLimiter limiter; + // Compression for compressed readers + public CompressionMetadata compression; public Builder(ChannelProxy channel) { this.channel = channel; - this.overrideLength = -1L; this.bufferSize = DEFAULT_BUFFER_SIZE; this.bufferType = BufferType.OFF_HEAP; - this.regions = null; - this.limiter = null; } /** The buffer size is typically already page aligned but if that is not the case @@ -400,38 +338,30 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa * buffer size unless we are throttling, in which case we may as well read the maximum * directly since the intention is to read the full file, see CASSANDRA-8630. * */ - private void setBufferSize() + private int adjustedBufferSize() { if (limiter != null) - { - bufferSize = MAX_BUFFER_SIZE; - return; - } + return MAX_BUFFER_SIZE; - if ((bufferSize & ~4095) != bufferSize) - { // should already be a page size multiple but if that's not case round it up - bufferSize = (bufferSize + 4095) & ~4095; - } - - bufferSize = Math.min(MAX_BUFFER_SIZE, bufferSize); + // should already be a page size multiple but if that's not case round it up + int wholePageSize = (bufferSize + 4095) & ~4095; + return Math.min(MAX_BUFFER_SIZE, wholePageSize); } - protected ByteBuffer createBuffer() + protected Rebufferer createRebufferer() { - setBufferSize(); - - buffer = regions == null - ? allocateBuffer(bufferSize, bufferType) - : regions.floor(0).buffer.duplicate(); - - buffer.limit(0); - return buffer; + return instantiateRebufferer(chunkReader(), limiter); } - public Builder overrideLength(long overrideLength) + public RebuffererFactory chunkReader() { - this.overrideLength = overrideLength; - return this; + if (compression != null) + return CompressedSegmentedFile.chunkReader(channel, compression, regions); + if (regions != null) + return new MmapRebufferer(channel, -1, regions); + + int adjustedSize = adjustedBufferSize(); + return new SimpleChunkReader(channel, -1, bufferType, adjustedSize); } public Builder bufferSize(int bufferSize) @@ -455,6 +385,12 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa return this; } + public Builder compression(CompressionMetadata metadata) + { + this.compression = metadata; + return this; + } + public Builder limiter(RateLimiter limiter) { this.limiter = limiter; @@ -463,12 +399,12 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa public RandomAccessReader build() { - return new RandomAccessReader(this); + return new RandomAccessReader(createRebufferer()); } public RandomAccessReader buildWithChannel() { - return new RandomAccessReaderWithOwnChannel(this); + return new RandomAccessReaderWithOwnChannel(createRebufferer()); } } @@ -479,9 +415,9 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa // not have a shared channel. public static class RandomAccessReaderWithOwnChannel extends RandomAccessReader { - protected RandomAccessReaderWithOwnChannel(Builder builder) + protected RandomAccessReaderWithOwnChannel(Rebufferer rebufferer) { - super(builder); + super(rebufferer); } @Override @@ -493,7 +429,14 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa } finally { - channel.close(); + try + { + rebufferer.close(); + } + finally + { + getChannel().close(); + } } } } diff --git a/src/java/org/apache/cassandra/io/util/ReaderFileProxy.java b/src/java/org/apache/cassandra/io/util/ReaderFileProxy.java new file mode 100644 index 0000000000..3ddb1437c7 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/ReaderFileProxy.java @@ -0,0 +1,36 @@ +/* + * 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; + +/** + * Base class for the RandomAccessReader components that implement reading. + */ +public interface ReaderFileProxy extends AutoCloseable +{ + void close(); // no checked exceptions + + ChannelProxy channel(); + + long fileLength(); + + /** + * Needed for tests. Returns the table's CRC check chance, which is only set for compressed tables. + */ + double getCrcCheckChance(); +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/util/Rebufferer.java b/src/java/org/apache/cassandra/io/util/Rebufferer.java new file mode 100644 index 0000000000..e88c7cbf35 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/Rebufferer.java @@ -0,0 +1,84 @@ +/* + * 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.ByteBuffer; + +/** + * Rebufferer for reading data by a RandomAccessReader. + */ +public interface Rebufferer extends ReaderFileProxy +{ + /** + * Rebuffer (move on or seek to) a given position, and return a buffer that can be used there. + * The only guarantee about the size of the returned data is that unless rebuffering at the end of the file, + * the buffer will not be empty and will contain the requested position, i.e. + * {@code offset <= position < offset + bh.buffer().limit()}, but the buffer will not be positioned there. + */ + BufferHolder rebuffer(long position); + + /** + * Called when a reader is closed. Should clean up reader-specific data. + */ + void closeReader(); + + public interface BufferHolder + { + /** + * Returns a useable buffer (i.e. one whose position and limit can be freely modified). Its limit will be set + * to the size of the available data in the buffer. + * The buffer must be treated as read-only. + */ + ByteBuffer buffer(); + + /** + * Position in the file of the start of the buffer. + */ + long offset(); + + /** + * To be called when this buffer is no longer in use. Must be called for all BufferHolders, or ChunkCache + * will not be able to free blocks. + */ + void release(); + } + + static final BufferHolder EMPTY = new BufferHolder() + { + final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0); + + @Override + public ByteBuffer buffer() + { + return EMPTY_BUFFER; + } + + @Override + public long offset() + { + return 0; + } + + @Override + public void release() + { + // nothing to do + } + }; +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/util/RebuffererFactory.java b/src/java/org/apache/cassandra/io/util/RebuffererFactory.java new file mode 100644 index 0000000000..ec35f0ba53 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/RebuffererFactory.java @@ -0,0 +1,32 @@ +/* + * 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; + +/** + * Interface for the classes that can be used to instantiate rebufferers over a given file. + * + * These are one of two types: + * - chunk sources (e.g. SimpleReadRebufferer) which instantiate a buffer managing rebufferer referencing + * themselves. + * - thread-safe shared rebufferers (e.g. MmapRebufferer) which directly return themselves. + */ +public interface RebuffererFactory extends ReaderFileProxy +{ + Rebufferer instantiateRebufferer(); +} diff --git a/src/java/org/apache/cassandra/io/util/SegmentedFile.java b/src/java/org/apache/cassandra/io/util/SegmentedFile.java index ab2d291bf1..62e14ba235 100644 --- a/src/java/org/apache/cassandra/io/util/SegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/SegmentedFile.java @@ -21,7 +21,6 @@ import java.io.DataInput; import java.io.DataOutput; import java.io.File; import java.io.IOException; -import java.util.function.Supplier; import com.google.common.util.concurrent.RateLimiter; @@ -52,27 +51,21 @@ import static org.apache.cassandra.utils.Throwables.maybeFail; public abstract class SegmentedFile extends SharedCloseableImpl { public final ChannelProxy channel; - public final int bufferSize; - public final long length; // This differs from length for compressed files (but we still need length for // SegmentIterator because offsets in the file are relative to the uncompressed size) public final long onDiskLength; /** - * Use getBuilder to get a Builder to construct a SegmentedFile. + * Rebufferer to use to construct RandomAccessReaders. */ - SegmentedFile(Cleanup cleanup, ChannelProxy channel, int bufferSize, long length) - { - this(cleanup, channel, bufferSize, length, length); - } + private final RebuffererFactory rebufferer; - protected SegmentedFile(Cleanup cleanup, ChannelProxy channel, int bufferSize, long length, long onDiskLength) + protected SegmentedFile(Cleanup cleanup, ChannelProxy channel, RebuffererFactory rebufferer, long onDiskLength) { super(cleanup); + this.rebufferer = rebufferer; this.channel = channel; - this.bufferSize = bufferSize; - this.length = length; this.onDiskLength = onDiskLength; } @@ -80,8 +73,7 @@ public abstract class SegmentedFile extends SharedCloseableImpl { super(copy); channel = copy.channel; - bufferSize = copy.bufferSize; - length = copy.length; + rebufferer = copy.rebufferer; onDiskLength = copy.onDiskLength; } @@ -90,12 +82,24 @@ public abstract class SegmentedFile extends SharedCloseableImpl return channel.filePath(); } + public long dataLength() + { + return rebufferer.fileLength(); + } + + public RebuffererFactory rebuffererFactory() + { + return rebufferer; + } + protected static class Cleanup implements RefCounted.Tidy { final ChannelProxy channel; - protected Cleanup(ChannelProxy channel) + final ReaderFileProxy rebufferer; + protected Cleanup(ChannelProxy channel, ReaderFileProxy rebufferer) { this.channel = channel; + this.rebufferer = rebufferer; } public String name() @@ -105,7 +109,14 @@ public abstract class SegmentedFile extends SharedCloseableImpl public void tidy() { - channel.close(); + try + { + channel.close(); + } + finally + { + rebufferer.close(); + } } } @@ -113,19 +124,12 @@ public abstract class SegmentedFile extends SharedCloseableImpl public RandomAccessReader createReader() { - return new RandomAccessReader.Builder(channel) - .overrideLength(length) - .bufferSize(bufferSize) - .build(); + return RandomAccessReader.build(this, null); } public RandomAccessReader createReader(RateLimiter limiter) { - return new RandomAccessReader.Builder(channel) - .overrideLength(length) - .bufferSize(bufferSize) - .limiter(limiter) - .build(); + return RandomAccessReader.build(this, limiter); } public FileDataInput createReader(long position) @@ -308,7 +312,7 @@ public abstract class SegmentedFile extends SharedCloseableImpl @Override public String toString() { return getClass().getSimpleName() + "(path='" + path() + '\'' + - ", length=" + length + + ", length=" + rebufferer.fileLength() + ')'; -} + } } diff --git a/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java b/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java new file mode 100644 index 0000000000..7bfb57b6b7 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java @@ -0,0 +1,78 @@ +/* + * 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.ByteBuffer; + +import org.apache.cassandra.io.compress.BufferType; + +class SimpleChunkReader extends AbstractReaderFileProxy implements ChunkReader +{ + private final int bufferSize; + private final BufferType bufferType; + + public SimpleChunkReader(ChannelProxy channel, long fileLength, BufferType bufferType, int bufferSize) + { + super(channel, fileLength); + this.bufferSize = bufferSize; + this.bufferType = bufferType; + } + + @Override + public void readChunk(long position, ByteBuffer buffer) + { + buffer.clear(); + channel.read(buffer, position); + buffer.flip(); + } + + @Override + public int chunkSize() + { + return bufferSize; + } + + @Override + public BufferType preferredBufferType() + { + return bufferType; + } + + @Override + public boolean alignmentRequired() + { + return false; + } + + @Override + public Rebufferer instantiateRebufferer() + { + return BufferManagingRebufferer.on(this); + } + + @Override + public String toString() + { + return String.format("%s(%s - chunk length %d, data length %d)", + getClass().getSimpleName(), + channel.filePath(), + bufferSize, + fileLength()); + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/metrics/CacheMetrics.java b/src/java/org/apache/cassandra/metrics/CacheMetrics.java index 151268befb..e623dcbd14 100644 --- a/src/java/org/apache/cassandra/metrics/CacheMetrics.java +++ b/src/java/org/apache/cassandra/metrics/CacheMetrics.java @@ -17,8 +17,6 @@ */ package org.apache.cassandra.metrics; -import java.util.concurrent.atomic.AtomicLong; - import com.codahale.metrics.Gauge; import com.codahale.metrics.Meter; import com.codahale.metrics.RatioGauge; @@ -56,7 +54,7 @@ public class CacheMetrics * @param type Type of Cache to identify metrics. * @param cache Cache to measure metrics */ - public CacheMetrics(String type, final ICache cache) + public CacheMetrics(String type, final ICache cache) { MetricNameFactory factory = new DefaultNameFactory("Cache", type); diff --git a/src/java/org/apache/cassandra/metrics/CacheMissMetrics.java b/src/java/org/apache/cassandra/metrics/CacheMissMetrics.java new file mode 100644 index 0000000000..19d61ef8f6 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/CacheMissMetrics.java @@ -0,0 +1,114 @@ +/* + * 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.metrics; + +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Meter; +import com.codahale.metrics.RatioGauge; +import com.codahale.metrics.Timer; +import org.apache.cassandra.cache.CacheSize; + +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +/** + * Metrics for {@code ICache}. + */ +public class CacheMissMetrics +{ + /** Cache capacity in bytes */ + public final Gauge capacity; + /** Total number of cache hits */ + public final Meter misses; + /** Total number of cache requests */ + public final Meter requests; + /** Latency of misses */ + public final Timer missLatency; + /** all time cache hit rate */ + public final Gauge hitRate; + /** 1m hit rate */ + public final Gauge oneMinuteHitRate; + /** 5m hit rate */ + public final Gauge fiveMinuteHitRate; + /** 15m hit rate */ + public final Gauge fifteenMinuteHitRate; + /** Total size of cache, in bytes */ + public final Gauge size; + /** Total number of cache entries */ + public final Gauge entries; + + /** + * Create metrics for given cache. + * + * @param type Type of Cache to identify metrics. + * @param cache Cache to measure metrics + */ + public CacheMissMetrics(String type, final CacheSize cache) + { + MetricNameFactory factory = new DefaultNameFactory("Cache", type); + + capacity = Metrics.register(factory.createMetricName("Capacity"), (Gauge) cache::capacity); + misses = Metrics.meter(factory.createMetricName("Misses")); + requests = Metrics.meter(factory.createMetricName("Requests")); + missLatency = Metrics.timer(factory.createMetricName("MissLatency")); + hitRate = Metrics.register(factory.createMetricName("HitRate"), new RatioGauge() + { + @Override + public Ratio getRatio() + { + long req = requests.getCount(); + long mis = misses.getCount(); + return Ratio.of(req - mis, req); + } + }); + oneMinuteHitRate = Metrics.register(factory.createMetricName("OneMinuteHitRate"), new RatioGauge() + { + protected Ratio getRatio() + { + double req = requests.getOneMinuteRate(); + double mis = misses.getOneMinuteRate(); + return Ratio.of(req - mis, req); + } + }); + fiveMinuteHitRate = Metrics.register(factory.createMetricName("FiveMinuteHitRate"), new RatioGauge() + { + protected Ratio getRatio() + { + double req = requests.getFiveMinuteRate(); + double mis = misses.getFiveMinuteRate(); + return Ratio.of(req - mis, req); + } + }); + fifteenMinuteHitRate = Metrics.register(factory.createMetricName("FifteenMinuteHitRate"), new RatioGauge() + { + protected Ratio getRatio() + { + double req = requests.getFifteenMinuteRate(); + double mis = misses.getFifteenMinuteRate(); + return Ratio.of(req - mis, req); + } + }); + size = Metrics.register(factory.createMetricName("Size"), (Gauge) cache::weightedSize); + entries = Metrics.register(factory.createMetricName("Entries"), (Gauge) cache::size); + } + + public void reset() + { + requests.mark(-requests.getCount()); + misses.mark(-misses.getCount()); + } +} diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java index 55ac7acc92..c7d5f98f90 100644 --- a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java +++ b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java @@ -60,7 +60,7 @@ public class CompressedInputStream extends InputStream // number of bytes in the buffer that are actually valid protected int validBufferBytes = -1; - private final Checksum checksum; + private final ChecksumType checksumType; // raw checksum bytes private final byte[] checksumBytes = new byte[4]; @@ -76,11 +76,11 @@ public class CompressedInputStream extends InputStream public CompressedInputStream(InputStream source, CompressionInfo info, ChecksumType checksumType, Supplier crcCheckChanceSupplier) { this.info = info; - this.checksum = checksumType.newInstance(); this.buffer = new byte[info.parameters.chunkLength()]; // buffer is limited to store up to 1024 chunks this.dataBuffer = new ArrayBlockingQueue<>(Math.min(info.chunks.length, 1024)); this.crcCheckChanceSupplier = crcCheckChanceSupplier; + this.checksumType = checksumType; new Thread(new Reader(source, info, dataBuffer)).start(); } @@ -122,14 +122,11 @@ public class CompressedInputStream extends InputStream // validate crc randomly if (this.crcCheckChanceSupplier.get() > ThreadLocalRandom.current().nextDouble()) { - checksum.update(compressed, 0, compressed.length - checksumBytes.length); + int checksum = (int) checksumType.of(compressed, 0, compressed.length - checksumBytes.length); System.arraycopy(compressed, compressed.length - checksumBytes.length, checksumBytes, 0, checksumBytes.length); - if (Ints.fromByteArray(checksumBytes) != (int) checksum.getValue()) + if (Ints.fromByteArray(checksumBytes) != checksum) throw new IOException("CRC unmatched"); - - // reset checksum object back to the original (blank) state - checksum.reset(); } // buffer offset is always aligned diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index fb6e1fb5e2..21183080e7 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -1179,9 +1179,18 @@ public class NodeProbe implements AutoCloseable CassandraMetricsRegistry.JmxGaugeMBean.class).getValue(); case "Requests": case "Hits": + case "Misses": return JMX.newMBeanProxy(mbeanServerConn, new ObjectName("org.apache.cassandra.metrics:type=Cache,scope=" + cacheType + ",name=" + metricName), CassandraMetricsRegistry.JmxMeterMBean.class).getCount(); + case "MissLatency": + return JMX.newMBeanProxy(mbeanServerConn, + new ObjectName("org.apache.cassandra.metrics:type=Cache,scope=" + cacheType + ",name=" + metricName), + CassandraMetricsRegistry.JmxTimerMBean.class).getMean(); + case "MissLatencyUnit": + return JMX.newMBeanProxy(mbeanServerConn, + new ObjectName("org.apache.cassandra.metrics:type=Cache,scope=" + cacheType + ",name=MissLatency"), + CassandraMetricsRegistry.JmxTimerMBean.class).getDurationUnit(); default: throw new RuntimeException("Unknown cache metric name."); diff --git a/src/java/org/apache/cassandra/tools/nodetool/Info.java b/src/java/org/apache/cassandra/tools/nodetool/Info.java index 268d9df9cd..c37f3b8e56 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/Info.java +++ b/src/java/org/apache/cassandra/tools/nodetool/Info.java @@ -117,6 +117,28 @@ public class Info extends NodeToolCmd probe.getCacheMetric("CounterCache", "HitRate"), cacheService.getCounterCacheSavePeriodInSeconds()); + // Chunk Cache: Hits, Requests, RecentHitRate, SavePeriodInSeconds + try + { + System.out.printf("%-23s: entries %d, size %s, capacity %s, %d misses, %d requests, %.3f recent hit rate, %.3f %s miss latency%n", + "Chunk Cache", + probe.getCacheMetric("ChunkCache", "Entries"), + FileUtils.stringifyFileSize((long) probe.getCacheMetric("ChunkCache", "Size")), + FileUtils.stringifyFileSize((long) probe.getCacheMetric("ChunkCache", "Capacity")), + probe.getCacheMetric("ChunkCache", "Misses"), + probe.getCacheMetric("ChunkCache", "Requests"), + probe.getCacheMetric("ChunkCache", "HitRate"), + probe.getCacheMetric("ChunkCache", "MissLatency"), + probe.getCacheMetric("ChunkCache", "MissLatencyUnit")); + } + catch (RuntimeException e) + { + if (!(e.getCause() instanceof InstanceNotFoundException)) + throw e; + + // Chunk cache is not on. + } + // check if node is already joined, before getting tokens, since it throws exception if not. if (probe.isJoined()) { diff --git a/src/java/org/apache/cassandra/utils/ChecksumType.java b/src/java/org/apache/cassandra/utils/ChecksumType.java index c9a1eb8164..3fa245b7cd 100644 --- a/src/java/org/apache/cassandra/utils/ChecksumType.java +++ b/src/java/org/apache/cassandra/utils/ChecksumType.java @@ -24,7 +24,7 @@ import java.util.zip.Adler32; public enum ChecksumType { - Adler32() + Adler32 { @Override @@ -40,7 +40,7 @@ public enum ChecksumType } }, - CRC32() + CRC32 { @Override @@ -58,6 +58,28 @@ public enum ChecksumType }; public abstract Checksum newInstance(); - public abstract void update(Checksum checksum, ByteBuffer buf); + + private ThreadLocal instances = ThreadLocal.withInitial(this::newInstance); + + public Checksum threadLocalInstance() + { + return instances.get(); + } + + public long of(ByteBuffer buf) + { + Checksum checksum = instances.get(); + checksum.reset(); + update(checksum, buf); + return checksum.getValue(); + } + + public long of(byte[] data, int off, int len) + { + Checksum checksum = instances.get(); + checksum.reset(); + checksum.update(data, off, len); + return checksum.getValue(); + } } diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java index 38c008dc22..ad2404f1ec 100644 --- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java +++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java @@ -46,7 +46,7 @@ import org.apache.cassandra.utils.concurrent.Ref; public class BufferPool { /** The size of a page aligned buffer, 64KiB */ - static final int CHUNK_SIZE = 64 << 10; + public static final int CHUNK_SIZE = 64 << 10; @VisibleForTesting public static long MEMORY_USAGE_THRESHOLD = DatabaseDescriptor.getFileCacheSizeInMB() * 1024L * 1024L; diff --git a/test/long/org/apache/cassandra/cql3/CachingBench.java b/test/long/org/apache/cassandra/cql3/CachingBench.java new file mode 100644 index 0000000000..370b3ffaf4 --- /dev/null +++ b/test/long/org/apache/cassandra/cql3/CachingBench.java @@ -0,0 +1,375 @@ +/* + * 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.cql3; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Random; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Predicate; + +import com.google.common.collect.Iterables; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import junit.framework.Assert; +import org.apache.cassandra.config.Config.CommitLogSync; +import org.apache.cassandra.config.Config.DiskAccessMode; +import org.apache.cassandra.cache.ChunkCache; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.db.rows.Row; +import org.apache.cassandra.db.rows.Unfiltered; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.io.sstable.ISSTableScanner; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.utils.FBUtilities; + +public class CachingBench extends CQLTester +{ + private static final String STRATEGY = "LeveledCompactionStrategy"; + + private static final int DEL_SECTIONS = 1000; + private static final int FLUSH_FREQ = 10000; + private static final int SCAN_FREQUENCY_INV = 12000; + static final int COUNT = 29000; + static final int ITERS = 9; + + static final int KEY_RANGE = 30; + static final int CLUSTERING_RANGE = 210000; + + static final int EXTRA_SIZE = 1025; + static final boolean CONCURRENT_COMPACTIONS = true; + + // The name of this method is important! + // CommitLog settings must be applied before CQLTester sets up; by using the same name as its @BeforeClass method we + // are effectively overriding it. + @BeforeClass + public static void setUpClass() + { + DatabaseDescriptor.setCommitLogSync(CommitLogSync.periodic); + DatabaseDescriptor.setCommitLogSyncPeriod(100); + CQLTester.setUpClass(); + } + + String hashQuery; + + @Before + public void before() throws Throwable + { + createTable("CREATE TABLE %s(" + + " key int," + + " column int," + + " data int," + + " extra text," + + " PRIMARY KEY(key, column)" + + ")" + ); + + String hashIFunc = parseFunctionName(createFunction(KEYSPACE, "int, int", + " CREATE FUNCTION %s (state int, val int)" + + " CALLED ON NULL INPUT" + + " RETURNS int" + + " LANGUAGE java" + + " AS 'return val != null ? state * 17 + val : state;'")).name; + String hashTFunc = parseFunctionName(createFunction(KEYSPACE, "int, text", + " CREATE FUNCTION %s (state int, val text)" + + " CALLED ON NULL INPUT" + + " RETURNS int" + + " LANGUAGE java" + + " AS 'return val != null ? state * 17 + val.hashCode() : state;'")).name; + + String hashInt = createAggregate(KEYSPACE, "int", + " CREATE AGGREGATE %s (int)" + + " SFUNC " + hashIFunc + + " STYPE int" + + " INITCOND 1"); + String hashText = createAggregate(KEYSPACE, "text", + " CREATE AGGREGATE %s (text)" + + " SFUNC " + hashTFunc + + " STYPE int" + + " INITCOND 1"); + + hashQuery = String.format("SELECT count(column), %s(key), %s(column), %s(data), %s(extra), avg(key), avg(column), avg(data) FROM %%s", + hashInt, hashInt, hashInt, hashText); + } + AtomicLong id = new AtomicLong(); + long compactionTimeNanos = 0; + + void pushData(Random rand, int count) throws Throwable + { + for (int i = 0; i < count; ++i) + { + long ii = id.incrementAndGet(); + if (ii % 1000 == 0) + System.out.print('.'); + int key = rand.nextInt(KEY_RANGE); + int column = rand.nextInt(CLUSTERING_RANGE); + execute("INSERT INTO %s (key, column, data, extra) VALUES (?, ?, ?, ?)", key, column, (int) ii, genExtra(rand)); + maybeCompact(ii); + } + } + + private String genExtra(Random rand) + { + StringBuilder builder = new StringBuilder(EXTRA_SIZE); + for (int i = 0; i < EXTRA_SIZE; ++i) + builder.append((char) ('a' + rand.nextInt('z' - 'a' + 1))); + return builder.toString(); + } + + void readAndDelete(Random rand, int count) throws Throwable + { + for (int i = 0; i < count; ++i) + { + int key; + UntypedResultSet res; + long ii = id.incrementAndGet(); + if (ii % 1000 == 0) + System.out.print('-'); + if (rand.nextInt(SCAN_FREQUENCY_INV) != 1) + { + do + { + key = rand.nextInt(KEY_RANGE); + long cid = rand.nextInt(DEL_SECTIONS); + int cstart = (int) (cid * CLUSTERING_RANGE / DEL_SECTIONS); + int cend = (int) ((cid + 1) * CLUSTERING_RANGE / DEL_SECTIONS); + res = execute("SELECT column FROM %s WHERE key = ? AND column >= ? AND column < ? LIMIT 1", key, cstart, cend); + } while (res.size() == 0); + UntypedResultSet.Row r = Iterables.get(res, rand.nextInt(res.size())); + int clustering = r.getInt("column"); + execute("DELETE FROM %s WHERE key = ? AND column = ?", key, clustering); + } + else + { + execute(hashQuery); + } + maybeCompact(ii); + } + } + + private void maybeCompact(long ii) + { + if (ii % FLUSH_FREQ == 0) + { + System.out.print("F"); + flush(); + if (ii % (FLUSH_FREQ * 10) == 0) + { + System.out.println("C"); + long startTime = System.nanoTime(); + getCurrentColumnFamilyStore().enableAutoCompaction(!CONCURRENT_COMPACTIONS); + long endTime = System.nanoTime(); + compactionTimeNanos += endTime - startTime; + getCurrentColumnFamilyStore().disableAutoCompaction(); + } + } + } + + public void testSetup(String compactionClass, String compressorClass, DiskAccessMode mode, boolean cacheEnabled) throws Throwable + { + id.set(0); + compactionTimeNanos = 0; + ChunkCache.instance.enable(cacheEnabled); + DatabaseDescriptor.setDiskAccessMode(mode); + alterTable("ALTER TABLE %s WITH compaction = { 'class' : '" + compactionClass + "' };"); + alterTable("ALTER TABLE %s WITH compression = { 'sstable_compression' : '" + compressorClass + "' };"); + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + cfs.disableAutoCompaction(); + + long onStartTime = System.currentTimeMillis(); + ExecutorService es = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); + List> tasks = new ArrayList<>(); + for (int ti = 0; ti < 1; ++ti) + { + Random rand = new Random(ti); + tasks.add(es.submit(() -> + { + for (int i = 0; i < ITERS; ++i) + try + { + pushData(rand, COUNT); + readAndDelete(rand, COUNT / 3); + } + catch (Throwable e) + { + throw new AssertionError(e); + } + })); + } + for (Future task : tasks) + task.get(); + + flush(); + long onEndTime = System.currentTimeMillis(); + int startRowCount = countRows(cfs); + int startTombCount = countTombstoneMarkers(cfs); + int startRowDeletions = countRowDeletions(cfs); + int startTableCount = cfs.getLiveSSTables().size(); + long startSize = SSTableReader.getTotalBytes(cfs.getLiveSSTables()); + System.out.println("\nCompession: " + cfs.getCompressionParameters().toString()); + System.out.println("Reader " + cfs.getLiveSSTables().iterator().next().getFileDataInput(0).toString()); + if (cacheEnabled) + System.out.format("Cache size %s requests %,d hit ratio %f\n", + FileUtils.stringifyFileSize(ChunkCache.instance.metrics.size.getValue()), + ChunkCache.instance.metrics.requests.getCount(), + ChunkCache.instance.metrics.hitRate.getValue()); + else + { + Assert.assertTrue("Chunk cache had requests: " + ChunkCache.instance.metrics.requests.getCount(), ChunkCache.instance.metrics.requests.getCount() < COUNT); + System.out.println("Cache disabled"); + } + System.out.println(String.format("Operations completed in %.3fs", (onEndTime - onStartTime) * 1e-3)); + if (!CONCURRENT_COMPACTIONS) + System.out.println(String.format(", out of which %.3f for non-concurrent compaction", compactionTimeNanos * 1e-9)); + else + System.out.println(); + + String hashesBefore = getHashes(); + long startTime = System.currentTimeMillis(); + CompactionManager.instance.performMaximal(cfs, true); + long endTime = System.currentTimeMillis(); + + int endRowCount = countRows(cfs); + int endTombCount = countTombstoneMarkers(cfs); + int endRowDeletions = countRowDeletions(cfs); + int endTableCount = cfs.getLiveSSTables().size(); + long endSize = SSTableReader.getTotalBytes(cfs.getLiveSSTables()); + + System.out.println(String.format("Major compaction completed in %.3fs", + (endTime - startTime) * 1e-3)); + System.out.println(String.format("At start: %,12d tables %12s %,12d rows %,12d deleted rows %,12d tombstone markers", + startTableCount, FileUtils.stringifyFileSize(startSize), startRowCount, startRowDeletions, startTombCount)); + System.out.println(String.format("At end: %,12d tables %12s %,12d rows %,12d deleted rows %,12d tombstone markers", + endTableCount, FileUtils.stringifyFileSize(endSize), endRowCount, endRowDeletions, endTombCount)); + String hashesAfter = getHashes(); + + Assert.assertEquals(hashesBefore, hashesAfter); + } + + private String getHashes() throws Throwable + { + long startTime = System.currentTimeMillis(); + String hashes = Arrays.toString(getRows(execute(hashQuery))[0]); + long endTime = System.currentTimeMillis(); + System.out.println(String.format("Hashes: %s, retrieved in %.3fs", hashes, (endTime - startTime) * 1e-3)); + return hashes; + } + + @Test + public void testWarmup() throws Throwable + { + testSetup(STRATEGY, "LZ4Compressor", DiskAccessMode.mmap, false); + } + + @Test + public void testLZ4CachedMmap() throws Throwable + { + testSetup(STRATEGY, "LZ4Compressor", DiskAccessMode.mmap, true); + } + + @Test + public void testLZ4CachedStandard() throws Throwable + { + testSetup(STRATEGY, "LZ4Compressor", DiskAccessMode.standard, true); + } + + @Test + public void testLZ4UncachedMmap() throws Throwable + { + testSetup(STRATEGY, "LZ4Compressor", DiskAccessMode.mmap, false); + } + + @Test + public void testLZ4UncachedStandard() throws Throwable + { + testSetup(STRATEGY, "LZ4Compressor", DiskAccessMode.standard, false); + } + + @Test + public void testCachedStandard() throws Throwable + { + testSetup(STRATEGY, "", DiskAccessMode.standard, true); + } + + @Test + public void testUncachedStandard() throws Throwable + { + testSetup(STRATEGY, "", DiskAccessMode.standard, false); + } + + @Test + public void testMmapped() throws Throwable + { + testSetup(STRATEGY, "", DiskAccessMode.mmap, false /* doesn't matter */); + } + + int countTombstoneMarkers(ColumnFamilyStore cfs) + { + return count(cfs, x -> x.isRangeTombstoneMarker()); + } + + int countRowDeletions(ColumnFamilyStore cfs) + { + return count(cfs, x -> x.isRow() && !((Row) x).deletion().isLive()); + } + + int countRows(ColumnFamilyStore cfs) + { + int nowInSec = FBUtilities.nowInSeconds(); + return count(cfs, x -> x.isRow() && ((Row) x).hasLiveData(nowInSec)); + } + + private int count(ColumnFamilyStore cfs, Predicate predicate) + { + int count = 0; + for (SSTableReader reader : cfs.getLiveSSTables()) + count += count(reader, predicate); + return count; + } + + int count(SSTableReader reader, Predicate predicate) + { + int instances = 0; + try (ISSTableScanner partitions = reader.getScanner()) + { + while (partitions.hasNext()) + { + try (UnfilteredRowIterator iter = partitions.next()) + { + while (iter.hasNext()) + { + Unfiltered atom = iter.next(); + if (predicate.test(atom)) + ++instances; + } + } + } + } + return instances; + } +} diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index 1b36c03f00..1e7d05f290 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java @@ -108,9 +108,6 @@ public abstract class CQLTester } PROTOCOL_VERSIONS = builder.build(); - // Once per-JVM is enough - prepareServer(); - nativeAddr = InetAddress.getLoopbackAddress(); try @@ -230,6 +227,9 @@ public abstract class CQLTester DatabaseDescriptor.setRowCacheSizeInMB(ROW_CACHE_SIZE_IN_MB); StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance); + + // Once per-JVM is enough + prepareServer(); } @AfterClass diff --git a/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java b/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java index 3c83da7f85..b6f8adacc5 100644 --- a/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java +++ b/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java @@ -34,6 +34,8 @@ public class SelectionColumnMappingTest extends CQLTester public static void setUpClass() { DatabaseDescriptor.setPartitionerUnsafe(ByteOrderedPartitioner.instance); + + prepareServer(); } @Test diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java index 4c7c135773..4c52ed26c9 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java @@ -41,10 +41,12 @@ import static org.junit.Assert.assertEquals; public class FrozenCollectionsTest extends CQLTester { @BeforeClass - public static void setUpClass() + public static void setUpClass() // overrides CQLTester.setUpClass() { // Selecting partitioner for a table is not exposed on CREATE TABLE. StorageService.instance.setPartitionerUnsafe(ByteOrderedPartitioner.instance); + + prepareServer(); } @Test diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java index 5501561e2c..2bffb09206 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java @@ -29,10 +29,12 @@ import org.apache.cassandra.service.StorageService; public class UserTypesTest extends CQLTester { @BeforeClass - public static void setUpClass() + public static void setUpClass() // overrides CQLTester.setUpClass() { // Selecting partitioner for a table is not exposed on CREATE TABLE. StorageService.instance.setPartitionerUnsafe(ByteOrderedPartitioner.instance); + + prepareServer(); } @Test diff --git a/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/CrcCheckChanceTest.java b/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/CrcCheckChanceTest.java index d059f7d16b..2760ae5800 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/CrcCheckChanceTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/CrcCheckChanceTest.java @@ -30,8 +30,8 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.compaction.CompactionInterruptedException; import org.apache.cassandra.db.compaction.CompactionManager; -import org.apache.cassandra.io.compress.CompressedRandomAccessReader; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.util.RandomAccessReader; import org.apache.cassandra.utils.FBUtilities; @@ -153,8 +153,8 @@ public class CrcCheckChanceTest extends CQLTester // note: only compressed files currently perform crc checks, so only the dfile reader is relevant here SSTableReader baseSSTable = cfs.getLiveSSTables().iterator().next(); SSTableReader idxSSTable = indexCfs.getLiveSSTables().iterator().next(); - try (CompressedRandomAccessReader baseDataReader = (CompressedRandomAccessReader)baseSSTable.openDataReader(); - CompressedRandomAccessReader idxDataReader = (CompressedRandomAccessReader)idxSSTable.openDataReader()) + try (RandomAccessReader baseDataReader = baseSSTable.openDataReader(); + RandomAccessReader idxDataReader = idxSSTable.openDataReader()) { Assert.assertEquals(0.03, baseDataReader.getCrcCheckChance()); Assert.assertEquals(0.03, idxDataReader.getCrcCheckChance()); diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java index a736668453..df654bfea4 100644 --- a/test/unit/org/apache/cassandra/db/ScrubTest.java +++ b/test/unit/org/apache/cassandra/db/ScrubTest.java @@ -29,6 +29,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.apache.cassandra.*; +import org.apache.cassandra.cache.ChunkCache; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.Operator; @@ -417,6 +418,8 @@ public class ScrubTest file.seek(startPosition); file.writeBytes(StringUtils.repeat('z', (int) (endPosition - startPosition))); file.close(); + if (ChunkCache.instance != null) + ChunkCache.instance.invalidateFile(sstable.getFilename()); } private static void assertOrderedAll(ColumnFamilyStore cfs, int expectedSize) diff --git a/test/unit/org/apache/cassandra/db/VerifyTest.java b/test/unit/org/apache/cassandra/db/VerifyTest.java index 9de01c1391..d216860c8c 100644 --- a/test/unit/org/apache/cassandra/db/VerifyTest.java +++ b/test/unit/org/apache/cassandra/db/VerifyTest.java @@ -23,6 +23,7 @@ import com.google.common.base.Charsets; import org.apache.cassandra.OrderedJUnit4ClassRunner; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; +import org.apache.cassandra.cache.ChunkCache; import org.apache.cassandra.UpdateBuilder; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.compaction.Verifier; @@ -313,6 +314,8 @@ public class VerifyTest file.seek(startPosition); file.writeBytes(StringUtils.repeat('z', (int) 2)); file.close(); + if (ChunkCache.instance != null) + ChunkCache.instance.invalidateFile(sstable.getFilename()); // Update the Digest to have the right Checksum writeChecksum(simpleFullChecksum(sstable.getFilename()), sstable.descriptor.filenameFor(sstable.descriptor.digestComponent)); diff --git a/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java index 19d2347b2a..b442df0921 100644 --- a/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java @@ -33,6 +33,7 @@ import static org.junit.Assert.assertNotNull; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; +import org.apache.cassandra.cache.ChunkCache; import org.apache.cassandra.db.*; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.sstable.format.SSTableReader; @@ -148,6 +149,8 @@ public class BlacklistingCompactionsTest byte[] corruption = new byte[corruptionSize]; Arrays.fill(corruption, (byte)0xFF); raf.write(corruption); + if (ChunkCache.instance != null) + ChunkCache.instance.invalidateFile(sstable.getFilename()); } finally diff --git a/test/unit/org/apache/cassandra/hints/ChecksummedDataInputTest.java b/test/unit/org/apache/cassandra/hints/ChecksummedDataInputTest.java index 323a12d603..6b3b5c8dc2 100644 --- a/test/unit/org/apache/cassandra/hints/ChecksummedDataInputTest.java +++ b/test/unit/org/apache/cassandra/hints/ChecksummedDataInputTest.java @@ -111,7 +111,7 @@ public class ChecksummedDataInputTest // assert that the crc matches, and that we've read exactly as many bytes as expected assertTrue(reader.checkCrc()); - assertEquals(0, reader.bytesRemaining()); + assertTrue(reader.isEOF()); reader.checkLimit(0); } @@ -177,7 +177,7 @@ public class ChecksummedDataInputTest assertEquals(2.2f, reader.readFloat()); assertEquals(42, reader.readInt()); assertTrue(reader.checkCrc()); - assertEquals(0, reader.bytesRemaining()); + assertTrue(reader.isEOF()); } } @@ -227,7 +227,7 @@ public class ChecksummedDataInputTest assertEquals(10, reader.readByte()); assertEquals('t', reader.readChar()); assertFalse(reader.checkCrc()); - assertEquals(0, reader.bytesRemaining()); + assertTrue(reader.isEOF()); } } } diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java index 74b5c7474b..53535e5351 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java @@ -96,8 +96,8 @@ public class CompressedRandomAccessReaderTest writer.finish(); } - try(RandomAccessReader reader = new CompressedRandomAccessReader.Builder(channel, - new CompressionMetadata(filename + ".metadata", f.length(), ChecksumType.CRC32)) + try(RandomAccessReader reader = RandomAccessReader.builder(channel) + .compression(new CompressionMetadata(filename + ".metadata", f.length(), ChecksumType.CRC32)) .build()) { String res = reader.readLine(); @@ -142,16 +142,17 @@ public class CompressedRandomAccessReaderTest assert f.exists(); CompressionMetadata compressionMetadata = compressed ? new CompressionMetadata(filename + ".metadata", f.length(), ChecksumType.CRC32) : null; - RandomAccessReader.Builder builder = compressed - ? new CompressedRandomAccessReader.Builder(channel, compressionMetadata) - : new RandomAccessReader.Builder(channel); + RandomAccessReader.Builder builder = RandomAccessReader.builder(channel); + if (compressed) + builder.compression(compressionMetadata); + MmappedRegions regions = null; if (usemmap) { - if (compressed) - builder.regions(MmappedRegions.map(channel, compressionMetadata)); - else - builder.regions(MmappedRegions.map(channel, f.length())); + regions = compressed + ? MmappedRegions.map(channel, compressionMetadata) + : MmappedRegions.map(channel, f.length()); + builder.regions(regions); } try(RandomAccessReader reader = builder.build()) @@ -163,8 +164,8 @@ public class CompressedRandomAccessReaderTest assert new String(b).equals(expected) : "Expecting '" + expected + "', got '" + new String(b) + '\''; } - if (usemmap) - builder.regions.close(); + if (regions != null) + regions.close(); } finally { @@ -203,7 +204,7 @@ public class CompressedRandomAccessReaderTest CompressionMetadata meta = new CompressionMetadata(metadata.getPath(), file.length(), ChecksumType.CRC32); CompressionMetadata.Chunk chunk = meta.chunkFor(0); - try(RandomAccessReader reader = new CompressedRandomAccessReader.Builder(channel, meta).build()) + try(RandomAccessReader reader = RandomAccessReader.builder(channel).compression(meta).build()) {// read and verify compressed data assertEquals(CONTENT, reader.readLine()); @@ -228,7 +229,7 @@ public class CompressedRandomAccessReaderTest checksumModifier.write(random.nextInt()); SyncUtil.sync(checksumModifier); // making sure that change was synced with disk - try (final RandomAccessReader r = new CompressedRandomAccessReader.Builder(channel, meta).build()) + try (final RandomAccessReader r = RandomAccessReader.builder(channel).compression(meta).build()) { Throwable exception = null; try @@ -248,7 +249,7 @@ public class CompressedRandomAccessReaderTest // lets write original checksum and check if we can read data updateChecksum(checksumModifier, chunk.length, checksum); - try (RandomAccessReader cr = new CompressedRandomAccessReader.Builder(channel, meta).build()) + try (RandomAccessReader cr = RandomAccessReader.builder(channel).compression(meta).build()) { // read and verify compressed data assertEquals(CONTENT, cr.readLine()); diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java index 66a0e284ac..ad2e4b48a4 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java @@ -117,7 +117,7 @@ public class CompressedSequentialWriterTest extends SequentialWriterTest } assert f.exists(); - RandomAccessReader reader = new CompressedRandomAccessReader.Builder(channel, new CompressionMetadata(filename + ".metadata", f.length(), ChecksumType.CRC32)).build(); + RandomAccessReader reader = RandomAccessReader.builder(channel).compression(new CompressionMetadata(filename + ".metadata", f.length(), ChecksumType.CRC32)).build(); assertEquals(dataPre.length + rawPost.length, reader.length()); byte[] result = new byte[(int)reader.length()]; diff --git a/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java b/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java index 360d262ce6..4ecbdcce9e 100644 --- a/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java +++ b/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java @@ -18,6 +18,7 @@ * */ package org.apache.cassandra.io.util; +import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.SyncUtil; @@ -330,7 +331,7 @@ public class BufferedRandomAccessFileTest { File file1 = writeTemporaryFile(new byte[16]); try (final ChannelProxy channel = new ChannelProxy(file1); - final RandomAccessReader file = new RandomAccessReader.Builder(channel) + final RandomAccessReader file = RandomAccessReader.builder(channel) .bufferSize(bufferSize) .build()) { @@ -343,7 +344,7 @@ public class BufferedRandomAccessFileTest { File file1 = writeTemporaryFile(new byte[16]); try (final ChannelProxy channel = new ChannelProxy(file1); - final RandomAccessReader file = new RandomAccessReader.Builder(channel).bufferSize(bufferSize).build()) + final RandomAccessReader file = RandomAccessReader.builder(channel).bufferSize(bufferSize).build()) { expectEOF(() -> { while (true) diff --git a/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java b/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java index 9df3fed468..85e8460498 100644 --- a/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java +++ b/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java @@ -99,8 +99,8 @@ public class MmappedRegionsTest { MmappedRegions.Region region = regions.floor(i); assertNotNull(region); - assertEquals(0, region.bottom()); - assertEquals(1024, region.top()); + assertEquals(0, region.offset()); + assertEquals(1024, region.end()); } regions.extend(2048); @@ -110,13 +110,13 @@ public class MmappedRegionsTest assertNotNull(region); if (i < 1024) { - assertEquals(0, region.bottom()); - assertEquals(1024, region.top()); + assertEquals(0, region.offset()); + assertEquals(1024, region.end()); } else { - assertEquals(1024, region.bottom()); - assertEquals(2048, region.top()); + assertEquals(1024, region.offset()); + assertEquals(2048, region.end()); } } } @@ -141,8 +141,8 @@ public class MmappedRegionsTest { MmappedRegions.Region region = regions.floor(i); assertNotNull(region); - assertEquals(SIZE * (i / SIZE), region.bottom()); - assertEquals(SIZE + (SIZE * (i / SIZE)), region.top()); + assertEquals(SIZE * (i / SIZE), region.offset()); + assertEquals(SIZE + (SIZE * (i / SIZE)), region.end()); } } finally @@ -169,8 +169,8 @@ public class MmappedRegionsTest { MmappedRegions.Region region = regions.floor(i); assertNotNull(region); - assertEquals(SIZE * (i / SIZE), region.bottom()); - assertEquals(SIZE + (SIZE * (i / SIZE)), region.top()); + assertEquals(SIZE * (i / SIZE), region.offset()); + assertEquals(SIZE + (SIZE * (i / SIZE)), region.end()); } } finally @@ -209,8 +209,8 @@ public class MmappedRegionsTest { MmappedRegions.Region region = snapshot.floor(i); assertNotNull(region); - assertEquals(SIZE * (i / SIZE), region.bottom()); - assertEquals(SIZE + (SIZE * (i / SIZE)), region.top()); + assertEquals(SIZE * (i / SIZE), region.offset()); + assertEquals(SIZE + (SIZE * (i / SIZE)), region.end()); // check we can access the buffer assertNotNull(region.buffer.duplicate().getInt()); @@ -267,8 +267,8 @@ public class MmappedRegionsTest { MmappedRegions.Region region = regions.floor(i); assertNotNull(region); - assertEquals(0, region.bottom()); - assertEquals(4096, region.top()); + assertEquals(0, region.offset()); + assertEquals(4096, region.end()); } } } @@ -326,8 +326,8 @@ public class MmappedRegionsTest assertNotNull(compressedChunk); assertEquals(chunk.length + 4, compressedChunk.capacity()); - assertEquals(chunk.offset, region.bottom()); - assertEquals(chunk.offset + chunk.length + 4, region.top()); + assertEquals(chunk.offset, region.offset()); + assertEquals(chunk.offset + chunk.length + 4, region.end()); i += metadata.chunkLength(); } diff --git a/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java index e83c0151a9..14c549cfb1 100644 --- a/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java @@ -67,12 +67,6 @@ public class RandomAccessReaderTest this.maxSegmentSize = maxSegmentSize; return this; } - - public Parameters expected(byte[] expected) - { - this.expected = expected; - return this; - } } @Test @@ -108,6 +102,7 @@ public class RandomAccessReaderTest @Test public void testMultipleSegments() throws IOException { + // FIXME: This is the same as above. testReadFully(new Parameters(8192, 4096).mmappedRegions(true).maxSegmentSize(1024)); } @@ -119,7 +114,7 @@ public class RandomAccessReaderTest try(ChannelProxy channel = new ChannelProxy("abc", new FakeFileChannel(SIZE))) { - RandomAccessReader.Builder builder = new RandomAccessReader.Builder(channel) + RandomAccessReader.Builder builder = RandomAccessReader.builder(channel) .bufferType(params.bufferType) .bufferSize(params.bufferSize); @@ -270,11 +265,15 @@ public class RandomAccessReaderTest final File f = writeFile(params); try(ChannelProxy channel = new ChannelProxy(f)) { - RandomAccessReader.Builder builder = new RandomAccessReader.Builder(channel) + RandomAccessReader.Builder builder = RandomAccessReader.builder(channel) .bufferType(params.bufferType) .bufferSize(params.bufferSize); + MmappedRegions regions = null; if (params.mmappedRegions) - builder.regions(MmappedRegions.map(channel, f.length())); + { + regions = MmappedRegions.map(channel, f.length()); + builder.regions(regions); + } try(RandomAccessReader reader = builder.build()) { @@ -296,8 +295,8 @@ public class RandomAccessReaderTest assertEquals(0, reader.bytesRemaining()); } - if (builder.regions != null) - assertNull(builder.regions.close(null)); + if (regions != null) + assertNull(regions.close(null)); } } @@ -316,7 +315,7 @@ public class RandomAccessReaderTest assert f.exists(); try(ChannelProxy channel = new ChannelProxy(f); - RandomAccessReader reader = new RandomAccessReader.Builder(channel).build()) + RandomAccessReader reader = RandomAccessReader.builder(channel).build()) { assertEquals(f.getAbsolutePath(), reader.getPath()); assertEquals(expected.length(), reader.length()); @@ -346,7 +345,7 @@ public class RandomAccessReaderTest assert f.exists(); try(ChannelProxy channel = new ChannelProxy(f); - RandomAccessReader reader = new RandomAccessReader.Builder(channel).build()) + RandomAccessReader reader = RandomAccessReader.builder(channel).build()) { assertEquals(expected.length() * numIterations, reader.length()); @@ -428,7 +427,7 @@ public class RandomAccessReaderTest { final Runnable worker = () -> { - try(RandomAccessReader reader = new RandomAccessReader.Builder(channel).build()) + try(RandomAccessReader reader = RandomAccessReader.builder(channel).build()) { assertEquals(expected.length, reader.length());