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

Add uncompressed chunk cache for RandomAccessReader 

patch by Branimir Lambov; reviewed by Pavel Yaskevich for CASSANDRA-5863
This commit is contained in:
Branimir Lambov 2016-03-02 11:15:44 +02:00 committed by Pavel Yaskevich
parent 16e3507534
commit 30bb255ec9
57 changed files with 2395 additions and 861 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.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)

11
conf/cassandra.yaml
View File

@ -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

BIN
lib/caffeine-2.2.6.jar Normal file
View File

Binary file not shown.

202
lib/licenses/caffeine-2.2.6.txt Normal file
View File

@ -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.

14
src/java/org/apache/cassandra/cache/CacheSize.java vendored Normal file
View File

@ -0,0 +1,14 @@
package org.apache.cassandra.cache;
public interface CacheSize
{
long capacity();
void setCapacity(long capacity);
int size();
long weightedSize();
}

301
src/java/org/apache/cassandra/cache/ChunkCache.java vendored Normal file
View File

@ -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<ChunkCache.Key, ChunkCache.Buffer>, RemovalListener<ChunkCache.Key, ChunkCache.Buffer>, 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<Key, Buffer> 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);
}
}

10
src/java/org/apache/cassandra/cache/ICache.java vendored
View File

@ -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<K, V>
public interface ICache<K, V> 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<K, V>
public void remove(K key);
public int size();
public long weightedSize();
public void clear();
public Iterator<K> keyIterator();

2
src/java/org/apache/cassandra/db/compaction/Scrubber.java
View File

@ -443,7 +443,7 @@ public class Scrubber implements Closeable
}
catch (Exception e)
{
throw new RuntimeException();
throw new RuntimeException(e);
}
}
}

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

@ -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;
}
}

44
src/java/org/apache/cassandra/hints/CompressedChecksummedDataInput.java
View File

@ -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

36
src/java/org/apache/cassandra/hints/CompressedChecksummedDataInputBuilder.java
View File

@ -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;
}
}

51
src/java/org/apache/cassandra/hints/EncryptedChecksummedDataInput.java
View File

@ -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

10
src/java/org/apache/cassandra/hints/HintsReader.java
View File

@ -111,7 +111,7 @@ class HintsReader implements AutoCloseable, Iterable<HintsReader.Page>
void seek(long newPosition)
{
input.seek(newPosition);
throw new UnsupportedOperationException("Hints are not seekable.");
}
public Iterator<Page> iterator()
@ -149,12 +149,12 @@ class HintsReader implements AutoCloseable, Iterable<HintsReader.Page>
@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<HintsReader.Page>
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<HintsReader.Page>
do
{
long position = input.getFilePointer();
long position = input.getSourcePosition();
if (input.isEOF())
return endOfData(); // reached EOF

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

@ -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);
}
}
}

2
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -1637,7 +1637,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
*/
public long uncompressedLength()
{
return dfile.length;
return dfile.dataLength();
}
/**

13
src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
View File

@ -32,6 +32,8 @@ import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.cache.ChunkCache;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.rows.*;
@ -58,7 +60,9 @@ public class BigTableWriter extends SSTableWriter
private final SegmentedFile.Builder dbuilder;
protected final SequentialWriter dataFile;
private DecoratedKey lastWrittenKey;
private DataPosition dataMark;
private long lastEarlyOpenLength = 0;
public BigTableWriter(Descriptor descriptor,
Long keyCount,
@ -265,6 +269,7 @@ public class BigTableWriter extends SSTableWriter
IndexSummary indexSummary = iwriter.summary.build(metadata.partitioner, boundary);
SegmentedFile ifile = iwriter.builder.buildIndex(descriptor, indexSummary, boundary);
SegmentedFile dfile = dbuilder.buildData(descriptor, stats, boundary);
invalidateCacheAtBoundary(dfile);
SSTableReader sstable = SSTableReader.internalOpen(descriptor,
components, metadata,
ifile, dfile, indexSummary,
@ -276,6 +281,13 @@ public class BigTableWriter extends SSTableWriter
return sstable;
}
void invalidateCacheAtBoundary(SegmentedFile dfile)
{
if (ChunkCache.instance != null && lastEarlyOpenLength != 0 && dfile.dataLength() > 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,

61
src/java/org/apache/cassandra/io/util/AbstractReaderFileProxy.java Normal file
View File

@ -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.
}
}

127
src/java/org/apache/cassandra/io/util/BufferManagingRebufferer.java Normal file
View File

@ -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();
}
}
}

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

@ -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)

1
src/java/org/apache/cassandra/io/util/ChannelProxy.java
View File

@ -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)

111
src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java
View File

@ -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();
}
}
}

56
src/java/org/apache/cassandra/io/util/ChunkReader.java Normal file
View File

@ -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();
}

291
src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java
View File

@ -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<ByteBuffer> 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();
}
}
}

13
src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java
View File

@ -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);

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

@ -23,6 +23,4 @@ public interface ICompressedFile
{
ChannelProxy channel();
CompressionMetadata getMetadata();
MmappedRegions regions();
}

106
src/java/org/apache/cassandra/io/util/LimitingRebufferer.java Normal file
View File

@ -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();
}
}

49
src/java/org/apache/cassandra/io/util/MmapRebufferer.java Normal file
View File

@ -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());
}
}

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

@ -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);

60
src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java
View File

@ -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)

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

@ -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();
}
}
}
}

36
src/java/org/apache/cassandra/io/util/ReaderFileProxy.java Normal file
View File

@ -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();
}

84
src/java/org/apache/cassandra/io/util/Rebufferer.java Normal file
View File

@ -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
}
};
}

32
src/java/org/apache/cassandra/io/util/RebuffererFactory.java Normal file
View File

@ -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();
}

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

@ -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() +
')';
}
}
}

78
src/java/org/apache/cassandra/io/util/SimpleChunkReader.java Normal file
View File

@ -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());
}
}

4
src/java/org/apache/cassandra/metrics/CacheMetrics.java
View File

@ -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);

114
src/java/org/apache/cassandra/metrics/CacheMissMetrics.java Normal file
View File

@ -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<Long> 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<Double> hitRate;
/** 1m hit rate */
public final Gauge<Double> oneMinuteHitRate;
/** 5m hit rate */
public final Gauge<Double> fiveMinuteHitRate;
/** 15m hit rate */
public final Gauge<Double> fifteenMinuteHitRate;
/** Total size of cache, in bytes */
public final Gauge<Long> size;
/** Total number of cache entries */
public final Gauge<Integer> 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<Long>) 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<Long>) cache::weightedSize);
entries = Metrics.register(factory.createMetricName("Entries"), (Gauge<Integer>) cache::size);
}
public void reset()
{
requests.mark(-requests.getCount());
misses.mark(-misses.getCount());
}
}

11
src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
View File

@ -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<Double> 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

9
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -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.");

22
src/java/org/apache/cassandra/tools/nodetool/Info.java
View File

@ -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())
{

28
src/java/org/apache/cassandra/utils/ChecksumType.java
View File

@ -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<Checksum> 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();
}
}

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

@ -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;

375
test/long/org/apache/cassandra/cql3/CachingBench.java Normal file
View File

@ -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<Future<?>> 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<Unfiltered> predicate)
{
int count = 0;
for (SSTableReader reader : cfs.getLiveSSTables())
count += count(reader, predicate);
return count;
}
int count(SSTableReader reader, Predicate<Unfiltered> 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;
}
}

6
test/unit/org/apache/cassandra/cql3/CQLTester.java
View File

@ -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

2
test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java
View File

@ -34,6 +34,8 @@ public class SelectionColumnMappingTest extends CQLTester
public static void setUpClass()
{
DatabaseDescriptor.setPartitionerUnsafe(ByteOrderedPartitioner.instance);
prepareServer();
}
@Test

4
test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java
View File

@ -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

4
test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java
View File

@ -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

6
test/unit/org/apache/cassandra/cql3/validation/miscellaneous/CrcCheckChanceTest.java
View File

@ -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());

3
test/unit/org/apache/cassandra/db/ScrubTest.java
View File

@ -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)

3
test/unit/org/apache/cassandra/db/VerifyTest.java
View File

@ -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));

3
test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
View File

@ -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

6
test/unit/org/apache/cassandra/hints/ChecksummedDataInputTest.java
View File

@ -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());
}
}
}

29
test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java
View File

@ -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());

2
test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java
View File

@ -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()];

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

@ -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)

32
test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java
View File

@ -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();
}

27
test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java
View File

@ -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());