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

add vInt encoding to Data(Input|Output)Plus 

patch by ariel and benedict for CASSANDRA-9499
This commit is contained in:
Ariel Weisberg 2015-06-15 14:31:03 -04:00 committed by Benedict Elliott Smith
parent 6092b01e32
commit 1491a40b7b
14 changed files with 633 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -10,7 +10,7 @@
* Change default garbage collector to G1 (CASSANDRA-7486)
* Populate TokenMetadata early during startup (CASSANDRA-9317)
* undeprecate cache recentHitRate (CASSANDRA-6591)
* Add support for selectively varint encoding fields (CASSANDRA-9499)
2.2.0-rc2
* (cqlsh) Allow setting the initial connection timeout (CASSANDRA-9601)

5
NOTICE.txt
View File

@ -74,3 +74,8 @@ OHC
(https://github.com/snazy/ohc)
Java Off-Heap-Cache, licensed under APLv2
Copyright 2014-2015 Robert Stupp, Germany.
Protocol buffers for varint encoding
https://developers.google.com/protocol-buffers/
Copyright 2008 Google Inc. All rights reserved.
BSD 3-clause

23
src/java/org/apache/cassandra/db/TypeSizes.java
View File

@ -20,6 +20,8 @@ package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import java.util.UUID;
import org.apache.cassandra.utils.vint.VIntCoding;
public abstract class TypeSizes
{
public static final TypeSizes NATIVE = new NativeDBTypeSizes();
@ -106,26 +108,7 @@ public abstract class TypeSizes
public int sizeofVInt(long i)
{
if (i >= -112 && i <= 127)
return 1;
int size = 0;
int len = -112;
if (i < 0)
{
i ^= -1L; // take one's complement'
len = -120;
}
long tmp = i;
while (tmp != 0)
{
tmp = tmp >> 8;
len--;
}
size++;
len = (len < -120) ? -(len + 120) : -(len + 112);
size += len;
return size;
return VIntCoding.computeVIntSize(i);
}
public int sizeof(long i)

34
src/java/org/apache/cassandra/io/util/AbstractDataInput.java
View File

@ -19,6 +19,8 @@ package org.apache.cassandra.io.util;
import java.io.*;
import org.apache.cassandra.utils.vint.VIntCoding;
public abstract class AbstractDataInput extends InputStream implements DataInput
{
public abstract void seek(long position) throws IOException;
@ -264,6 +266,38 @@ public abstract class AbstractDataInput extends InputStream implements DataInput
return ((long)(readInt()) << 32) + (readInt() & 0xFFFFFFFFL);
}
/**
* Reads a varint encoded integer from the current position in this file. Blocks until
* the end of the varint is reached, the end of the file is reached, or an exception is
* thrown.
*
* @return the next varint value from this file.
* @throws EOFException
* if the end of this file is detected.
* @throws IOException
* if this file is closed or another I/O error occurs.
*/
public long readVInt() throws IOException
{
return VIntCoding.readVInt(this);
}
/**
* Reads an unsigned varint encoded integer from the current position in this file. Blocks until
* the end of the varint is reached, the end of the file is reached, or an exception is
* thrown.
*
* @return the next unsigned varint value from this file.
* @throws EOFException
* if the end of this file is detected.
* @throws IOException
* if this file is closed or another I/O error occurs.
*/
public long readUnsignedVInt() throws IOException
{
return VIntCoding.readUnsignedVInt(this);
}
/**
* Reads a 16-bit short from the current position in this file. Blocks until
* two bytes have been read, the end of the file is reached or an exception

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

@ -29,7 +29,7 @@ import com.google.common.base.Preconditions;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.apache.cassandra.utils.vint.VIntCoding;
/**
* An implementation of the DataOutputStreamPlus interface using a ByteBuffer to stage writes
@ -213,6 +213,27 @@ public class BufferedDataOutputStreamPlus extends DataOutputStreamPlus
buffer.putLong(v);
}
@Override
public void writeVInt(long value) throws IOException
{
writeUnsignedVInt(VIntCoding.encodeZigZag64(value));
}
@Override
public void writeUnsignedVInt(long value) throws IOException
{
int size = VIntCoding.computeUnsignedVIntSize(value);
if (size == 1)
{
ensureRemaining(1);
buffer.put((byte) value);
return;
}
ensureRemaining(size);
buffer.put(VIntCoding.encodeVInt(value, size), 0, size);
}
@Override
public void writeFloat(float v) throws IOException
{

19
src/java/org/apache/cassandra/io/util/DataOutputPlus.java
View File

@ -22,6 +22,8 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import org.apache.cassandra.utils.vint.VIntCoding;
import com.google.common.base.Function;
/**
@ -40,4 +42,21 @@ public interface DataOutputPlus extends DataOutput
* and forget to flush
*/
<R> R applyToChannel(Function<WritableByteChannel, R> c) throws IOException;
default void writeVInt(long i) throws IOException
{
VIntCoding.writeVInt(i, this);
}
/**
* Think hard before opting for an unsigned encoding. Is this going to bite someone because some day
* they might need to pass in a sentinel value using negative numbers? Is the risk worth it
* to save a few bytes?
*
* Signed, not a fan of unsigned values in protocols and formats
*/
default void writeUnsignedVInt(long i) throws IOException
{
VIntCoding.writeUnsignedVInt(i, this);
}
}

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

@ -27,6 +27,8 @@ import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SeekableByteChannel;
import org.apache.cassandra.utils.vint.VIntCoding;
import com.google.common.base.Preconditions;
/**
@ -50,7 +52,7 @@ public class NIODataInputStream extends InputStream implements DataInput, Closea
public NIODataInputStream(ReadableByteChannel rbc, int bufferSize)
{
Preconditions.checkNotNull(rbc);
Preconditions.checkArgument(bufferSize >= 8, "Buffer size must be large enough to accomadate a long/double");
Preconditions.checkArgument(bufferSize >= 9, "Buffer size must be large enough to accomadate a varint");
this.rbc = rbc;
buf = ByteBuffer.allocateDirect(bufferSize);
buf.position(0);
@ -116,7 +118,7 @@ public class NIODataInputStream extends InputStream implements DataInput, Closea
private int readNext() throws IOException
{
Preconditions.checkState(buf.remaining() != buf.capacity());
assert(buf.remaining() < 8);
assert(buf.remaining() < 9);
/*
* If there is data already at the start of the buffer, move the position to the end
@ -151,31 +153,55 @@ public class NIODataInputStream extends InputStream implements DataInput, Closea
return read;
}
/*
* Read at least minimum bytes and throw EOF if that fails
*/
private void readMinimum(int minimum) throws IOException
/*
* Read at least minimum bytes and throw EOF if that fails
*/
private void readMinimum(int attempt, int require) throws IOException
{
assert(buf.remaining() < 8);
while (buf.remaining() < minimum)
int remaining;
while ((remaining = buf.remaining()) < attempt)
{
int read = readNext();
if (read == -1)
{
//DataInputStream consumes the bytes even if it doesn't get the entire value, match the behavior here
buf.position(0);
buf.limit(0);
throw new EOFException();
if (remaining < require)
{
//DataInputStream consumes the bytes even if it doesn't get the entire value, match the behavior here
buf.position(0);
buf.limit(0);
throw new EOFException();
}
}
}
}
/*
* Ensure the buffer contains the minimum number of readable bytes
* Ensure the buffer contains the minimum number of readable bytes, throws EOF if enough bytes aren't available
* Add padding if requested and return the limit of the buffer without any padding that is added.
*/
private int prepareReadPaddedPrimitive(int minimum) throws IOException
{
int limitToSet = buf.limit();
int position = buf.position();
if (limitToSet - position < minimum)
{
readMinimum(minimum, 1);
limitToSet = buf.limit();
position = buf.position();
if (limitToSet - position < minimum)
buf.limit(position + minimum);
}
return limitToSet;
}
/*
* Ensure the buffer contains the minimum number of readable bytes, throws EOF if enough bytes aren't available
*/
private void prepareReadPrimitive(int minimum) throws IOException
{
if (buf.remaining() < minimum) readMinimum(minimum);
if (buf.remaining() < minimum)
readMinimum(minimum, minimum);
}
@Override
@ -248,6 +274,40 @@ public class NIODataInputStream extends InputStream implements DataInput, Closea
return buf.getLong();
}
public long readVInt() throws IOException
{
return VIntCoding.decodeZigZag64(readUnsignedVInt());
}
public long readUnsignedVInt() throws IOException
{
byte firstByte = readByte();
//Bail out early if this is one byte, necessary or it fails later
if (firstByte >= 0)
return firstByte;
//If padding was added, the limit to set after to get rid of the padding
int limitToSet = prepareReadPaddedPrimitive(8);
int position = buf.position();
int extraBytes = VIntCoding.numberOfExtraBytesToRead(firstByte);
int extraBits = extraBytes * 8;
long retval = buf.getLong(position);
buf.position(position + extraBytes);
buf.limit(limitToSet);
// truncate the bytes we read in excess of those we needed
retval >>>= 64 - extraBits;
// remove the non-value bits from the first byte
firstByte &= VIntCoding.firstByteValueMask(extraBytes);
// shift the first byte up to its correct position
retval |= (long) firstByte << extraBits;
return retval;
}
@Override
public float readFloat() throws IOException
{

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

@ -23,7 +23,6 @@ import java.io.UTFDataFormatException;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.utils.memory.MemoryUtil;
import com.google.common.base.Function;

47
src/java/org/apache/cassandra/utils/vint/EncodedDataInputStream.java
View File

@ -27,6 +27,9 @@ import org.apache.cassandra.io.util.AbstractDataInput;
* https://developers.google.com/protocol-buffers/docs/encoding#varints
*
* Should be used with EncodedDataOutputStream
*
* @deprecated Where possible use NIODataInputStream which has a more efficient implementation of buffered input
* for most read methods
*/
public class EncodedDataInputStream extends AbstractDataInput implements DataInput
{
@ -71,55 +74,21 @@ public class EncodedDataInputStream extends AbstractDataInput implements DataInp
public int readInt() throws IOException
{
return (int) vintDecode();
return (int) VIntCoding.readVInt(input);
}
public long readLong() throws IOException
{
return vintDecode();
return VIntCoding.readVInt(input);
}
public int readUnsignedShort() throws IOException
{
return (short) vintDecode();
return (short) VIntCoding.readVInt(input);
}
public short readShort() throws IOException
{
return (short) vintDecode();
}
private long vintDecode() throws IOException
{
byte firstByte = input.readByte();
int len = vintDecodeSize(firstByte);
if (len == 1)
return firstByte;
long i = 0;
for (int idx = 0; idx < len - 1; idx++)
{
byte b = input.readByte();
i = i << 8;
i = i | (b & 0xFF);
}
return (vintIsNegative(firstByte) ? (i ^ -1L) : i);
}
private int vintDecodeSize(byte value)
{
if (value >= -112)
{
return 1;
}
else if (value < -120)
{
return -119 - value;
}
return -111 - value;
}
private boolean vintIsNegative(byte value)
{
return value < -120 || (value >= -112 && value < 0);
return (short) VIntCoding.readVInt(input);
}
}

35
src/java/org/apache/cassandra/utils/vint/EncodedDataOutputStream.java
View File

@ -54,45 +54,16 @@ public class EncodedDataOutputStream extends UnbufferedDataOutputStreamPlus
public void writeInt(int v) throws IOException
{
vintEncode(v);
writeVInt(v);
}
public void writeLong(long v) throws IOException
{
vintEncode(v);
writeVInt(v);
}
public void writeShort(int v) throws IOException
{
vintEncode(v);
}
private void vintEncode(long i) throws IOException
{
if (i >= -112 && i <= 127)
{
writeByte((byte) i);
return;
}
int len = -112;
if (i < 0)
{
i ^= -1L; // take one's complement'
len = -120;
}
long tmp = i;
while (tmp != 0)
{
tmp = tmp >> 8;
len--;
}
writeByte((byte) len);
len = (len < -120) ? -(len + 120) : -(len + 112);
for (int idx = len; idx != 0; idx--)
{
int shiftbits = (idx - 1) * 8;
long mask = 0xFFL << shiftbits;
writeByte((byte) ((i & mask) >> shiftbits));
}
writeVInt(v);
}
}

183
src/java/org/apache/cassandra/utils/vint/VIntCoding.java Normal file
View File

@ -0,0 +1,183 @@
/*
* 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.
*/
// Protocol Buffers - Google's data interchange format
// Copyright 2008 Google Inc. All rights reserved.
// https://developers.google.com/protocol-buffers/
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
// * Neither the name of Google Inc. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
package org.apache.cassandra.utils.vint;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import net.nicoulaj.compilecommand.annotations.Inline;
/**
* Borrows idea from
* https://developers.google.com/protocol-buffers/docs/encoding#varints
*/
public class VIntCoding
{
public static long readUnsignedVInt(DataInput input) throws IOException {
int firstByte = input.readByte();
//Bail out early if this is one byte, necessary or it fails later
if (firstByte >= 0)
return firstByte;
int size = numberOfExtraBytesToRead(firstByte);
long retval = firstByte & firstByteValueMask(size);;
for (int ii = 0; ii < size; ii++)
{
byte b = input.readByte();
retval <<= 8;
retval |= b & 0xff;
}
return retval;
}
public static long readVInt(DataInput input) throws IOException {
return decodeZigZag64(readUnsignedVInt(input));
}
// & this with the first byte to give the value part for a given extraBytesToRead encoded in the byte
public static int firstByteValueMask(int extraBytesToRead)
{
// by including the known 0bit in the mask, we can use this for encodeExtraBytesToRead
return 0xff >> extraBytesToRead;
}
public static int encodeExtraBytesToRead(int extraBytesToRead)
{
// because we have an extra bit in the value mask, we just need to invert it
return ~firstByteValueMask(extraBytesToRead);
}
public static int numberOfExtraBytesToRead(int firstByte)
{
// we count number of set upper bits; so if we simply invert all of the bits, we're golden
// this is aided by the fact that we only work with negative numbers, so when upcast to an int all
// of the new upper bits are also set, so by inverting we set all of them to zero
return Integer.numberOfLeadingZeros(~firstByte) - 24;
}
protected static final ThreadLocal<byte[]> encodingBuffer = new ThreadLocal<byte[]>()
{
@Override
public byte[] initialValue()
{
return new byte[9];
}
};
public static void writeUnsignedVInt(long value, DataOutput output) throws IOException {
int size = VIntCoding.computeUnsignedVIntSize(value);
if (size == 1)
{
output.write((int)value);
return;
}
output.write(VIntCoding.encodeVInt(value, size), 0, size);
}
@Inline
public static byte[] encodeVInt(long value, int size) {
byte encodingSpace[] = encodingBuffer.get();
int extraBytes = size - 1;
for (int i = extraBytes ; i >= 0; --i)
{
encodingSpace[i] = (byte) value;
value >>= 8;
}
encodingSpace[0] |= VIntCoding.encodeExtraBytesToRead(extraBytes);
return encodingSpace;
}
public static void writeVInt(long value, DataOutput output) throws IOException {
writeUnsignedVInt(encodeZigZag64(value), output);
}
/**
* Decode a ZigZag-encoded 64-bit value. ZigZag encodes signed integers
* into values that can be efficiently encoded with varint. (Otherwise,
* negative values must be sign-extended to 64 bits to be varint encoded,
* thus always taking 10 bytes on the wire.)
*
* @param n An unsigned 64-bit integer, stored in a signed int because
* Java has no explicit unsigned support.
* @return A signed 64-bit integer.
*/
public static long decodeZigZag64(final long n) {
return (n >>> 1) ^ -(n & 1);
}
/**
* Encode a ZigZag-encoded 64-bit value. ZigZag encodes signed integers
* into values that can be efficiently encoded with varint. (Otherwise,
* negative values must be sign-extended to 64 bits to be varint encoded,
* thus always taking 10 bytes on the wire.)
*
* @param n A signed 64-bit integer.
* @return An unsigned 64-bit integer, stored in a signed int because
* Java has no explicit unsigned support.
*/
public static long encodeZigZag64(final long n) {
// Note: the right-shift must be arithmetic
return (n << 1) ^ (n >> 63);
}
/** Compute the number of bytes that would be needed to encode a varint. */
public static int computeVIntSize(final long param) {
return computeUnsignedVIntSize(encodeZigZag64(param));
}
/** Compute the number of bytes that would be needed to encode an unsigned varint. */
public static int computeUnsignedVIntSize(final long value) {
int magnitude = Long.numberOfLeadingZeros(value | 1); // | with 1 to ensure magntiude <= 63, so (63 - 1) / 7 <= 8
return 9 - ((magnitude - 1) / 7);
}
}

95
test/unit/org/apache/cassandra/io/util/BufferedDataOutputStreamTest.java
View File

@ -8,10 +8,16 @@ import java.lang.reflect.Field;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import java.util.Arrays;
import java.util.Random;
import org.apache.cassandra.utils.vint.VIntCoding;
import org.junit.Test;
import com.google.common.primitives.UnsignedBytes;
import com.google.common.primitives.UnsignedInteger;
import com.google.common.primitives.UnsignedLong;
import static org.junit.Assert.*;
public class BufferedDataOutputStreamTest
@ -185,7 +191,7 @@ public class BufferedDataOutputStreamTest
int action = 0;
while (generated.size() < 1024 * 1024 * 8)
{
action = r.nextInt(19);
action = r.nextInt(21);
//System.out.println("Action " + action + " iteration " + iteration);
iteration++;
@ -362,6 +368,20 @@ public class BufferedDataOutputStreamTest
}
break;
}
case 19:
{
long val = r.nextLong();
VIntCoding.writeVInt(val, dosp);
ndosp.writeVInt(val);
break;
}
case 20:
{
long val = r.nextLong();
VIntCoding.writeUnsignedVInt(val, dosp);
ndosp.writeUnsignedVInt(val);
break;
}
default:
fail("Shouldn't reach here");
}
@ -442,4 +462,77 @@ public class BufferedDataOutputStreamTest
}
return count;
}
/*
* Add values to the array with a bit set in every position
*/
public static long[] enrich(long vals[])
{
long retval[] = Arrays.copyOf(vals, vals.length + 64);
for (int ii = 0; ii < 64; ii++)
retval[vals.length + ii] = 1L << ii;
return retval;
}
@Test
public void testVInt() throws Exception
{
setUp();
long testValues[] = new long[] {
0, 1, -1
,Long.MIN_VALUE, Long.MIN_VALUE + 1, Long.MAX_VALUE, Long.MAX_VALUE - 1
,Integer.MIN_VALUE, Integer.MIN_VALUE + 1, Integer.MAX_VALUE, Integer.MAX_VALUE - 1
,Short.MIN_VALUE, Short.MIN_VALUE + 1, Short.MAX_VALUE, Short.MAX_VALUE - 1
,Byte.MIN_VALUE, Byte.MIN_VALUE + 1, Byte.MAX_VALUE, Byte.MAX_VALUE - 1 };
testValues = enrich(testValues);
int expectedSize = 0;
for (long v : testValues)
{
expectedSize += VIntCoding.computeVIntSize(v);
ndosp.writeVInt(v);
}
ndosp.flush();
@SuppressWarnings("resource")
ByteBufferDataInput bbdi = new ByteBufferDataInput(ByteBuffer.wrap(generated.toByteArray()), "", 0, 0);
assertEquals(expectedSize, generated.toByteArray().length);
for (long v : testValues)
{
assertEquals(v, bbdi.readVInt());
}
}
@Test
public void testUnsignedVInt() throws Exception
{
setUp();
long testValues[] = new long[] { //-1 };
0, 1
, UnsignedLong.MAX_VALUE.longValue(), UnsignedLong.MAX_VALUE.longValue() - 1, UnsignedLong.MAX_VALUE.longValue() + 1
, UnsignedInteger.MAX_VALUE.longValue(), UnsignedInteger.MAX_VALUE.longValue() - 1, UnsignedInteger.MAX_VALUE.longValue() + 1
, UnsignedBytes.MAX_VALUE, UnsignedBytes.MAX_VALUE - 1, UnsignedBytes.MAX_VALUE + 1
, 65536, 65536 - 1, 65536 + 1 };
testValues = enrich(testValues);
int expectedSize = 0;
for (long v : testValues)
{
expectedSize += VIntCoding.computeUnsignedVIntSize(v);
ndosp.writeUnsignedVInt(v);
}
ndosp.flush();
@SuppressWarnings("resource")
ByteBufferDataInput bbdi = new ByteBufferDataInput(ByteBuffer.wrap(generated.toByteArray()), "", 0, 0);
assertEquals(expectedSize, generated.toByteArray().length);
for (long v : testValues)
assertEquals(v, bbdi.readUnsignedVInt());
}
}

122
test/unit/org/apache/cassandra/io/util/NIODataInputStreamTest.java
View File

@ -18,6 +18,9 @@ import org.apache.cassandra.io.util.NIODataInputStream;
import org.junit.Test;
import com.google.common.base.Charsets;
import com.google.common.primitives.UnsignedBytes;
import com.google.common.primitives.UnsignedInteger;
import com.google.common.primitives.UnsignedLong;
import static org.junit.Assert.*;
@ -126,7 +129,7 @@ public class NIODataInputStreamTest
}
NIODataInputStream fakeStream = new NIODataInputStream(new FakeChannel(), 8);
NIODataInputStream fakeStream = new NIODataInputStream(new FakeChannel(), 9);
@Test(expected = IOException.class)
public void testResetThrows() throws Exception
@ -212,7 +215,7 @@ public class NIODataInputStreamTest
fos.write(new byte[10]);
fos.seek(0);
is = new NIODataInputStream(fos.getChannel(), 8);
is = new NIODataInputStream(fos.getChannel(), 9);
int remaining = 10;
assertEquals(10, is.available());
@ -226,6 +229,31 @@ public class NIODataInputStreamTest
assertEquals(0, is.available());
}
private static ReadableByteChannel wrap(final byte bytes[])
{
final ByteBuffer buf = ByteBuffer.wrap(bytes);
return new ReadableByteChannel()
{
@Override
public boolean isOpen() {return false;}
@Override
public void close() throws IOException {}
@Override
public int read(ByteBuffer dst) throws IOException
{
int read = Math.min(dst.remaining(), buf.remaining());
buf.limit(buf.position() + read);
dst.put(buf);
buf.limit(buf.capacity());
return read == 0 ? -1 : read;
}
};
}
@SuppressWarnings("resource")
@Test
public void testReadUTF() throws Exception
@ -244,23 +272,7 @@ public class NIODataInputStreamTest
daos.writeUTF(BufferedDataOutputStreamTest.threeByte);
daos.writeUTF(BufferedDataOutputStreamTest.fourByte);
NIODataInputStream is = new NIODataInputStream(new ReadableByteChannel()
{
@Override
public boolean isOpen() {return false;}
@Override
public void close() throws IOException {}
@Override
public int read(ByteBuffer dst) throws IOException
{
dst.put(baos.toByteArray());
return baos.toByteArray().length;
}
}, 4096);
NIODataInputStream is = new NIODataInputStream(wrap(baos.toByteArray()), 4096);
assertEquals(simple, is.readUTF());
assertEquals(BufferedDataOutputStreamTest.twoByte, is.readUTF());
@ -268,6 +280,78 @@ public class NIODataInputStreamTest
assertEquals(BufferedDataOutputStreamTest.fourByte, is.readUTF());
}
@SuppressWarnings("resource")
@Test
public void testReadVInt() throws Exception {
final ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStreamPlus daos = new WrappedDataOutputStreamPlus(baos);
long values[] = new long[] {
0, 1, -1,
Long.MIN_VALUE, Long.MIN_VALUE + 1, Long.MAX_VALUE, Long.MAX_VALUE - 1,
Integer.MIN_VALUE, Integer.MIN_VALUE + 1, Integer.MAX_VALUE, Integer.MAX_VALUE - 1,
Short.MIN_VALUE, Short.MIN_VALUE + 1, Short.MAX_VALUE, Short.MAX_VALUE - 1,
Byte.MIN_VALUE, Byte.MIN_VALUE + 1, Byte.MAX_VALUE, Byte.MAX_VALUE - 1 };
values = BufferedDataOutputStreamTest.enrich(values);
for (long v : values)
daos.writeVInt(v);
daos.flush();
NIODataInputStream is = new NIODataInputStream(wrap(baos.toByteArray()), 9);
for (long v : values)
assertEquals(v, is.readVInt());
boolean threw = false;
try
{
is.readVInt();
}
catch (EOFException e)
{
threw = true;
}
assertTrue(threw);
}
@SuppressWarnings("resource")
@Test
public void testReadUnsignedVInt() throws Exception {
final ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStreamPlus daos = new WrappedDataOutputStreamPlus(baos);
long values[] = new long[] {
0, 1
, UnsignedLong.MAX_VALUE.longValue(), UnsignedLong.MAX_VALUE.longValue() - 1, UnsignedLong.MAX_VALUE.longValue() + 1
, UnsignedInteger.MAX_VALUE.longValue(), UnsignedInteger.MAX_VALUE.longValue() - 1, UnsignedInteger.MAX_VALUE.longValue() + 1
, UnsignedBytes.MAX_VALUE, UnsignedBytes.MAX_VALUE - 1, UnsignedBytes.MAX_VALUE + 1
, 65536, 65536 - 1, 65536 + 1 };
values = BufferedDataOutputStreamTest.enrich(values);
for (long v : values)
daos.writeUnsignedVInt(v);
daos.flush();
NIODataInputStream is = new NIODataInputStream(wrap(baos.toByteArray()), 9);
for (long v : values)
assertEquals(v, is.readUnsignedVInt());
boolean threw = false;
try
{
is.readUnsignedVInt();
}
catch (EOFException e)
{
threw = true;
}
assertTrue(threw);
}
@Test
public void testFuzz() throws Exception
{

85
test/unit/org/apache/cassandra/utils/vint/VIntCodingTest.java Normal file
View File

@ -0,0 +1,85 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.utils.vint;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.junit.Test;
import junit.framework.Assert;
public class VIntCodingTest
{
@Test
public void testComputeSize() throws Exception
{
assertEncodedAtExpectedSize(0L, 1);
for (int size = 1 ; size < 8 ; size++)
{
assertEncodedAtExpectedSize((1L << 7 * size) - 1, size);
assertEncodedAtExpectedSize(1L << 7 * size, size + 1);
}
Assert.assertEquals(9, VIntCoding.computeUnsignedVIntSize(Long.MAX_VALUE));
}
private void assertEncodedAtExpectedSize(long value, int expectedSize) throws Exception
{
Assert.assertEquals(expectedSize, VIntCoding.computeUnsignedVIntSize(value));
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos);
VIntCoding.writeUnsignedVInt(value, dos);
dos.flush();
Assert.assertEquals( expectedSize, baos.toByteArray().length);
DataOutputBuffer dob = new DataOutputBuffer();
dob.writeUnsignedVInt(value);
Assert.assertEquals( expectedSize, dob.buffer().remaining());
dob.close();
}
@Test
public void testReadExtraBytesCount()
{
for (int i = 1 ; i < 8 ; i++)
Assert.assertEquals(i, VIntCoding.numberOfExtraBytesToRead((byte) ((0xFF << (8 - i)) & 0xFF)));
}
/*
* Quick sanity check that 1 byte encodes up to 127 as expected
*/
@Test
public void testOneByteCapacity() throws Exception {
int biggestOneByte = 127;
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos);
VIntCoding.writeUnsignedVInt(biggestOneByte, dos);
dos.flush();
Assert.assertEquals( 1, baos.toByteArray().length);
DataOutputBuffer dob = new DataOutputBuffer();
dob.writeUnsignedVInt(biggestOneByte);
Assert.assertEquals( 1, dob.buffer().remaining());
dob.close();
}
}