You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2015/03/30 22:41:10 UTC
svn commit: r1670190 - in /hive/branches/llap:
ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/
serde/src/java/org/apache/hadoop/hive/serde2/
Author: sershe
Date: Mon Mar 30 20:41:10 2015
New Revision: 1670190
URL: http://svn.apache.org/r1670190
Log:
HIVE-10128 : LLAP: BytesBytesMultiHashMap does not allow concurrent read-only access (Sergey Shelukhin)
Modified:
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java?rev=1670190&r1=1670189&r2=1670190&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java Mon Mar 30 20:41:10 2015
@@ -184,6 +184,18 @@ public final class BytesBytesMultiHashMa
this(initialCapacity, loadFactor, wbSize, -1);
}
+ public class ThreadSafeGetter {
+ private WriteBuffers.Position position = new WriteBuffers.Position();
+ public byte getValueRefs(byte[] key, int length, List<WriteBuffers.ByteSegmentRef> result) {
+ return BytesBytesMultiHashMap.this.getValueRefs(key, length, result, position);
+ }
+
+ public void populateValue(WriteBuffers.ByteSegmentRef valueRef) {
+ // Convenience method, populateValue is thread-safe.
+ BytesBytesMultiHashMap.this.populateValue(valueRef);
+ }
+ }
+
/** The source of keys and values to put into hashtable; avoids byte copying. */
public static interface KvSource {
/** Write key into output. */
@@ -201,7 +213,7 @@ public final class BytesBytesMultiHashMa
}
/**
- * Adds new value to new or existing key in hashmap.
+ * Adds new value to new or existing key in hashmap. Not thread-safe.
* @param kv Keyvalue writer. Each method will be called at most once.
*/
private static final byte[] FOUR_ZEROES = new byte[] { 0, 0, 0, 0 };
@@ -247,6 +259,15 @@ public final class BytesBytesMultiHashMa
++numValues;
}
+ public ThreadSafeGetter createGetterForThread() {
+ return new ThreadSafeGetter();
+ }
+
+ /** Not thread-safe! Use createGetterForThread. */
+ public byte getValueRefs(byte[] key, int length, List<WriteBuffers.ByteSegmentRef> result) {
+ return getValueRefs(key, length, result, writeBuffers.getReadPosition());
+ }
+
/**
* Gets "lazy" values for a key (as a set of byte segments in underlying buffer).
* @param key Key buffer.
@@ -254,21 +275,21 @@ public final class BytesBytesMultiHashMa
* @param result The list to use to store the results.
* @return the state byte for the key (see class description).
*/
- public byte getValueRefs(byte[] key, int length, List<WriteBuffers.ByteSegmentRef> result) {
+ private byte getValueRefs(byte[] key, int length,
+ List<WriteBuffers.ByteSegmentRef> result, WriteBuffers.Position readPos) {
// First, find first record for the key.
result.clear();
- long ref = findKeyRefToRead(key, length);
+ long ref = findKeyRefToRead(key, length, readPos);
if (ref == 0) {
return 0;
}
boolean hasList = Ref.hasList(ref);
// This relies on findKeyRefToRead doing key equality check and leaving read ptr where needed.
- long lrPtrOffset = hasList ? writeBuffers.getReadPoint() : 0;
+ long lrPtrOffset = hasList ? writeBuffers.getReadPoint(readPos) : 0;
- writeBuffers.setReadPoint(getFirstRecordLengthsOffset(ref));
- int valueLength = (int)writeBuffers.readVLong();
- // LOG.info("Returning value at " + (Ref.getOffset(ref) - valueLength) + " length " + valueLength);
+ writeBuffers.setReadPoint(getFirstRecordLengthsOffset(ref, readPos), readPos);
+ int valueLength = (int)writeBuffers.readVLong(readPos);
result.add(new WriteBuffers.ByteSegmentRef(Ref.getOffset(ref) - valueLength, valueLength));
byte stateByte = Ref.getStateByte(ref);
if (!hasList) {
@@ -276,19 +297,17 @@ public final class BytesBytesMultiHashMa
}
// There're multiple records for the key; get the offset of the next one.
- long nextTailOffset = writeBuffers.readFiveByteULong(lrPtrOffset);
+ long nextTailOffset = writeBuffers.readNByteLong(lrPtrOffset, 5, readPos);
// LOG.info("Next tail offset " + nextTailOffset);
while (nextTailOffset > 0) {
- writeBuffers.setReadPoint(nextTailOffset);
- valueLength = (int)writeBuffers.readVLong();
- // LOG.info("Returning value at " + (nextTailOffset - valueLength) + " length " + valueLength);
+ writeBuffers.setReadPoint(nextTailOffset, readPos);
+ valueLength = (int)writeBuffers.readVLong(readPos);
result.add(new WriteBuffers.ByteSegmentRef(nextTailOffset - valueLength, valueLength));
// Now read the relative offset to next record. Next record is always before the
// previous record in the write buffers (see writeBuffers javadoc).
- long delta = writeBuffers.readVLong();
+ long delta = writeBuffers.readVLong(readPos);
nextTailOffset = delta == 0 ? 0 : (nextTailOffset - delta);
- // LOG.info("Delta " + delta + ", next tail offset " + nextTailOffset);
}
return stateByte;
}
@@ -388,7 +407,7 @@ public final class BytesBytesMultiHashMa
* @param length Read key length.
* @return The ref to use for reading.
*/
- private long findKeyRefToRead(byte[] key, int length) {
+ private long findKeyRefToRead(byte[] key, int length, WriteBuffers.Position readPos) {
final int bucketMask = (refs.length - 1);
int hashCode = writeBuffers.hashCode(key, 0, length);
int slot = hashCode & bucketMask;
@@ -402,7 +421,7 @@ public final class BytesBytesMultiHashMa
if (ref == 0) {
return 0;
}
- if (isSameKey(key, length, ref, hashCode)) {
+ if (isSameKey(key, length, ref, hashCode, readPos)) {
return ref;
}
probeSlot += (++i);
@@ -453,7 +472,7 @@ public final class BytesBytesMultiHashMa
if (!compareHashBits(ref, hashCode)) {
return false; // Hash bits in ref don't match.
}
- writeBuffers.setReadPoint(getFirstRecordLengthsOffset(ref));
+ writeBuffers.setReadPoint(getFirstRecordLengthsOffset(ref, null));
int valueLength = (int)writeBuffers.readVLong(), keyLength = (int)writeBuffers.readVLong();
if (keyLength != cmpLength) {
return false;
@@ -471,12 +490,14 @@ public final class BytesBytesMultiHashMa
/**
* Same as {@link #isSameKey(long, int, long, int)} but for externally stored key.
*/
- private boolean isSameKey(byte[] key, int length, long ref, int hashCode) {
+ private boolean isSameKey(byte[] key, int length, long ref, int hashCode,
+ WriteBuffers.Position readPos) {
if (!compareHashBits(ref, hashCode)) {
return false; // Hash bits don't match.
}
- writeBuffers.setReadPoint(getFirstRecordLengthsOffset(ref));
- int valueLength = (int)writeBuffers.readVLong(), keyLength = (int)writeBuffers.readVLong();
+ writeBuffers.setReadPoint(getFirstRecordLengthsOffset(ref, readPos), readPos);
+ int valueLength = (int)writeBuffers.readVLong(readPos),
+ keyLength = (int)writeBuffers.readVLong(readPos);
long keyOffset = Ref.getOffset(ref) - (valueLength + keyLength);
// See the comment in the other isSameKey
return writeBuffers.isEqual(key, length, keyOffset, keyLength);
@@ -491,10 +512,11 @@ public final class BytesBytesMultiHashMa
* @param ref Reference.
* @return The offset to value and key length vlongs of the first record referenced by ref.
*/
- private long getFirstRecordLengthsOffset(long ref) {
+ private long getFirstRecordLengthsOffset(long ref, WriteBuffers.Position readPos) {
long tailOffset = Ref.getOffset(ref);
if (Ref.hasList(ref)) {
- long relativeOffset = writeBuffers.readFiveByteULong(tailOffset);
+ long relativeOffset = (readPos == null) ? writeBuffers.readNByteLong(tailOffset, 5)
+ : writeBuffers.readNByteLong(tailOffset, 5, readPos);
tailOffset += relativeOffset;
}
return tailOffset;
@@ -522,10 +544,10 @@ public final class BytesBytesMultiHashMa
// TODO: we could actually store a bit flag in ref indicating whether this is a hash
// match or a probe, and in the former case use hash bits (for a first few resizes).
// int hashCodeOrPart = oldSlot | Ref.getNthHashBit(oldRef, startingHashBitCount, newHashBitCount);
- writeBuffers.setReadPoint(getFirstRecordLengthsOffset(oldRef));
+ writeBuffers.setReadPoint(getFirstRecordLengthsOffset(oldRef, null));
// Read the value and key length for the first record.
- int hashCode = writeBuffers.readInt(Ref.getOffset(oldRef)
- - writeBuffers.readVLong() - writeBuffers.readVLong() - 4);
+ int hashCode = (int)writeBuffers.readNByteLong(Ref.getOffset(oldRef)
+ - writeBuffers.readVLong() - writeBuffers.readVLong() - 4, 4);
int probeSteps = relocateKeyRef(newRefs, oldRef, hashCode);
maxSteps = Math.max(probeSteps, maxSteps);
}
@@ -576,7 +598,7 @@ public final class BytesBytesMultiHashMa
*/
private void addRecordToList(long lrPtrOffset, long tailOffset) {
// Now, insert this record into the list.
- long prevHeadOffset = writeBuffers.readFiveByteULong(lrPtrOffset);
+ long prevHeadOffset = writeBuffers.readNByteLong(lrPtrOffset, 5);
// LOG.info("Reading offset " + prevHeadOffset + " at " + lrPtrOffset);
assert prevHeadOffset < tailOffset; // We replace an earlier element, must have lower offset.
writeBuffers.writeFiveByteULong(lrPtrOffset, tailOffset);
@@ -632,7 +654,7 @@ public final class BytesBytesMultiHashMa
return tailOffset;
}
- /** Writes the debug dump of the table into logs. */
+ /** Writes the debug dump of the table into logs. Not thread-safe. */
public void debugDumpTable() {
StringBuilder dump = new StringBuilder(keysAssigned + " keys\n");
TreeMap<Long, Integer> byteIntervals = new TreeMap<Long, Integer>();
@@ -644,10 +666,11 @@ public final class BytesBytesMultiHashMa
continue;
}
++examined;
- long recOffset = getFirstRecordLengthsOffset(ref);
+ long recOffset = getFirstRecordLengthsOffset(ref, null);
long tailOffset = Ref.getOffset(ref);
writeBuffers.setReadPoint(recOffset);
- int valueLength = (int)writeBuffers.readVLong(), keyLength = (int)writeBuffers.readVLong();
+ int valueLength = (int)writeBuffers.readVLong(),
+ keyLength = (int)writeBuffers.readVLong();
long ptrOffset = writeBuffers.getReadPoint();
if (Ref.hasList(ref)) {
byteIntervals.put(recOffset, (int)(ptrOffset + 5 - recOffset));
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java?rev=1670190&r1=1670189&r2=1670190&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java Mon Mar 30 20:41:10 2015
@@ -430,6 +430,8 @@ public class MapJoinBytesTableContainer
private class ReusableRowContainer
implements MapJoinRowContainer, AbstractRowContainer.RowIterator<List<Object>> {
private byte aliasFilter;
+ /** Hash table wrapper; not really thread-local, just specific to the container. */
+ private final BytesBytesMultiHashMap.ThreadSafeGetter threadLocalHashMap;
private List<WriteBuffers.ByteSegmentRef> refs;
private int currentRow;
/**
@@ -449,22 +451,20 @@ public class MapJoinBytesTableContainer
valueStruct = null; // No rows?
}
uselessIndirection = new ByteArrayRef();
+ threadLocalHashMap = hashMap.createGetterForThread();
clearRows();
}
public JoinUtil.JoinResult setFromOutput(Output output) {
- synchronized (hashMap) {
- if (refs == null) {
- refs = new ArrayList<WriteBuffers.ByteSegmentRef>();
- }
- byte aliasFilter = hashMap.getValueRefs(output.getData(), output.getLength(), refs);
- this.aliasFilter = refs.isEmpty() ? (byte) 0xff : aliasFilter;
- this.dummyRow = null;
+ if (refs == null) {
+ refs = new ArrayList<WriteBuffers.ByteSegmentRef>();
}
+ byte aliasFilter = hashMap.getValueRefs(output.getData(), output.getLength(), refs);
+ this.aliasFilter = refs.isEmpty() ? (byte) 0xff : aliasFilter;
+ this.dummyRow = null;
if (refs.isEmpty()) {
return JoinUtil.JoinResult.NOMATCH;
- }
- else {
+ } else {
return JoinUtil.JoinResult.MATCH;
}
}
@@ -524,22 +524,20 @@ public class MapJoinBytesTableContainer
dummyRow = null;
return result;
}
- synchronized (hashMap) {
- if (currentRow < 0 || refs.size() < currentRow)
- throw new HiveException("No rows");
- if (refs.size() == currentRow)
- return null;
- WriteBuffers.ByteSegmentRef ref = refs.get(currentRow++);
- if (ref.getLength() == 0) {
- return EMPTY_LIST; // shortcut, 0 length means no fields
- }
- if (ref.getBytes() == null) {
- hashMap.populateValue(ref);
- }
- uselessIndirection.setData(ref.getBytes());
- valueStruct.init(uselessIndirection, (int) ref.getOffset(), ref.getLength());
- return valueStruct.getFieldsAsList(); // TODO: should we unset bytes after that?
- }
+ if (currentRow < 0 || refs.size() < currentRow)
+ throw new HiveException("No rows");
+ if (refs.size() == currentRow)
+ return null;
+ WriteBuffers.ByteSegmentRef ref = refs.get(currentRow++);
+ if (ref.getLength() == 0) {
+ return EMPTY_LIST; // shortcut, 0 length means no fields
+ }
+ if (ref.getBytes() == null) {
+ hashMap.populateValue(ref);
+ }
+ uselessIndirection.setData(ref.getBytes());
+ valueStruct.init(uselessIndirection, (int) ref.getOffset(), ref.getLength());
+ return valueStruct.getFieldsAsList(); // TODO: should we unset bytes after that?
}
@Override
Modified: hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java?rev=1670190&r1=1670189&r2=1670190&view=diff
==============================================================================
--- hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java (original)
+++ hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java Mon Mar 30 20:41:10 2015
@@ -41,72 +41,93 @@ public final class WriteBuffers implemen
private final long offsetMask;
private final long maxSize;
- private byte[] currentWriteBuffer;
- private int currentWriteBufferIndex;
- /** The offset in the last writeBuffer where the values are added */
- private int currentWriteOffset = 0;
-
- private byte[] currentReadBuffer = null;
- private int currentReadBufferIndex = 0;
- private int currentReadOffset = 0;
+ public static class Position {
+ private byte[] buffer = null;
+ private int bufferIndex = 0;
+ private int offset = 0;
+ public void clear() {
+ buffer = null;
+ bufferIndex = offset = -1;
+ }
+ }
+
+ Position writePos = new Position(); // Position where we'd write
+ Position defaultReadPos = new Position(); // Position where we'd read (by default).
+
public WriteBuffers(int wbSize, long maxSize) {
this.wbSize = Integer.bitCount(wbSize) == 1 ? wbSize : (Integer.highestOneBit(wbSize) << 1);
this.wbSizeLog2 = 31 - Integer.numberOfLeadingZeros(this.wbSize);
this.offsetMask = this.wbSize - 1;
this.maxSize = maxSize;
- currentWriteBufferIndex = -1;
+ writePos.bufferIndex = -1;
nextBufferToWrite();
}
public long readVLong() {
- ponderNextBufferToRead();
- byte firstByte = currentReadBuffer[currentReadOffset++];
+ return readVLong(defaultReadPos);
+ }
+
+ public long readVLong(Position readPos) {
+ ponderNextBufferToRead(readPos);
+ byte firstByte = readPos.buffer[readPos.offset++];
int length = (byte) WritableUtils.decodeVIntSize(firstByte) - 1;
if (length == 0) {
return firstByte;
}
long i = 0;
- if (isAllInOneReadBuffer(length)) {
+ if (isAllInOneReadBuffer(length, readPos)) {
for (int idx = 0; idx < length; idx++) {
- i = (i << 8) | (currentReadBuffer[currentReadOffset + idx] & 0xFF);
+ i = (i << 8) | (readPos.buffer[readPos.offset + idx] & 0xFF);
}
- currentReadOffset += length;
+ readPos.offset += length;
} else {
for (int idx = 0; idx < length; idx++) {
- i = (i << 8) | (readNextByte() & 0xFF);
+ i = (i << 8) | (readNextByte(readPos) & 0xFF);
}
}
return (WritableUtils.isNegativeVInt(firstByte) ? (i ^ -1L) : i);
}
public void skipVLong() {
- ponderNextBufferToRead();
- byte firstByte = currentReadBuffer[currentReadOffset++];
+ skipVLong(defaultReadPos);
+ }
+
+ public void skipVLong(Position readPos) {
+ ponderNextBufferToRead(readPos);
+ byte firstByte = readPos.buffer[readPos.offset++];
int length = (byte) WritableUtils.decodeVIntSize(firstByte);
if (length > 1) {
- currentReadOffset += (length - 1);
+ readPos.offset += (length - 1);
}
- int diff = currentReadOffset - wbSize;
+ int diff = readPos.offset - wbSize;
while (diff >= 0) {
- ++currentReadBufferIndex;
- currentReadBuffer = writeBuffers.get(currentReadBufferIndex);
- currentReadOffset = diff;
- diff = currentReadOffset - wbSize;
+ ++readPos.bufferIndex;
+ readPos.buffer = writeBuffers.get(readPos.bufferIndex);
+ readPos.offset = diff;
+ diff = readPos.offset - wbSize;
}
}
public void setReadPoint(long offset) {
- currentReadBufferIndex = getBufferIndex(offset);
- currentReadBuffer = writeBuffers.get(currentReadBufferIndex);
- currentReadOffset = getOffset(offset);
+ setReadPoint(offset, defaultReadPos);
+ }
+
+ public void setReadPoint(long offset, Position readPos) {
+ readPos.bufferIndex = getBufferIndex(offset);
+ readPos.buffer = writeBuffers.get(readPos.bufferIndex);
+ readPos.offset = getOffset(offset);
}
public int hashCode(long offset, int length) {
- setReadPoint(offset);
- if (isAllInOneReadBuffer(length)) {
- int result = murmurHash(currentReadBuffer, currentReadOffset, length);
- currentReadOffset += length;
+ return hashCode(offset, length, defaultReadPos);
+ }
+
+ public int hashCode(long offset, int length, Position readPos) {
+ setReadPoint(offset, readPos);
+ if (isAllInOneReadBuffer(length, readPos)) {
+ int result = murmurHash(readPos.buffer, readPos.offset, length);
+ readPos.offset += length;
return result;
}
@@ -114,26 +135,26 @@ public final class WriteBuffers implemen
byte[] bytes = new byte[length];
int destOffset = 0;
while (destOffset < length) {
- ponderNextBufferToRead();
- int toRead = Math.min(length - destOffset, wbSize - currentReadOffset);
- System.arraycopy(currentReadBuffer, currentReadOffset, bytes, destOffset, toRead);
- currentReadOffset += toRead;
+ ponderNextBufferToRead(readPos);
+ int toRead = Math.min(length - destOffset, wbSize - readPos.offset);
+ System.arraycopy(readPos.buffer, readPos.offset, bytes, destOffset, toRead);
+ readPos.offset += toRead;
destOffset += toRead;
}
return murmurHash(bytes, 0, bytes.length);
}
- private byte readNextByte() {
+ private byte readNextByte(Position readPos) {
// This method is inefficient. It's only used when something crosses buffer boundaries.
- ponderNextBufferToRead();
- return currentReadBuffer[currentReadOffset++];
+ ponderNextBufferToRead(readPos);
+ return readPos.buffer[readPos.offset++];
}
- private void ponderNextBufferToRead() {
- if (currentReadOffset >= wbSize) {
- ++currentReadBufferIndex;
- currentReadBuffer = writeBuffers.get(currentReadBufferIndex);
- currentReadOffset = 0;
+ private void ponderNextBufferToRead(Position readPos) {
+ if (readPos.offset >= wbSize) {
+ ++readPos.bufferIndex;
+ readPos.buffer = writeBuffers.get(readPos.bufferIndex);
+ readPos.offset = 0;
}
}
@@ -149,26 +170,26 @@ public final class WriteBuffers implemen
@Override
public void reserve(int byteCount) {
if (byteCount < 0) throw new AssertionError("byteCount must be non-negative");
- int currentWriteOffset = this.currentWriteOffset + byteCount;
+ int currentWriteOffset = writePos.offset + byteCount;
while (currentWriteOffset > wbSize) {
nextBufferToWrite();
currentWriteOffset -= wbSize;
}
- this.currentWriteOffset = currentWriteOffset;
+ writePos.offset = currentWriteOffset;
}
public void setWritePoint(long offset) {
- currentWriteBufferIndex = getBufferIndex(offset);
- currentWriteBuffer = writeBuffers.get(currentWriteBufferIndex);
- currentWriteOffset = getOffset(offset);
+ writePos.bufferIndex = getBufferIndex(offset);
+ writePos.buffer = writeBuffers.get(writePos.bufferIndex);
+ writePos.offset = getOffset(offset);
}
@Override
public void write(int b) {
- if (currentWriteOffset == wbSize) {
+ if (writePos.offset == wbSize) {
nextBufferToWrite();
}
- currentWriteBuffer[currentWriteOffset++] = (byte)b;
+ writePos.buffer[writePos.offset++] = (byte)b;
}
@Override
@@ -180,11 +201,11 @@ public final class WriteBuffers implemen
public void write(byte[] b, int off, int len) {
int srcOffset = 0;
while (srcOffset < len) {
- int toWrite = Math.min(len - srcOffset, wbSize - currentWriteOffset);
- System.arraycopy(b, srcOffset + off, currentWriteBuffer, currentWriteOffset, toWrite);
- currentWriteOffset += toWrite;
+ int toWrite = Math.min(len - srcOffset, wbSize - writePos.offset);
+ System.arraycopy(b, srcOffset + off, writePos.buffer, writePos.offset, toWrite);
+ writePos.offset += toWrite;
srcOffset += toWrite;
- if (currentWriteOffset == wbSize) {
+ if (writePos.offset == wbSize) {
nextBufferToWrite();
}
}
@@ -204,16 +225,16 @@ public final class WriteBuffers implemen
}
private void nextBufferToWrite() {
- if (currentWriteBufferIndex == (writeBuffers.size() - 1)) {
+ if (writePos.bufferIndex == (writeBuffers.size() - 1)) {
if ((1 + writeBuffers.size()) * ((long)wbSize) > maxSize) {
// We could verify precisely at write time, but just do approximate at allocation time.
throw new RuntimeException("Too much memory used by write buffers");
}
writeBuffers.add(new byte[wbSize]);
}
- ++currentWriteBufferIndex;
- currentWriteBuffer = writeBuffers.get(currentWriteBufferIndex);
- currentWriteOffset = 0;
+ ++writePos.bufferIndex;
+ writePos.buffer = writeBuffers.get(writePos.bufferIndex);
+ writePos.offset = 0;
}
/** Compares two parts of the buffer with each other. Does not modify readPoint. */
@@ -284,16 +305,25 @@ public final class WriteBuffers implemen
public void clear() {
writeBuffers.clear();
- currentWriteBuffer = currentReadBuffer = null;
- currentWriteOffset = currentReadOffset = currentWriteBufferIndex = currentReadBufferIndex = 0;
+ clearState();
}
+
+ private void clearState() {
+ writePos.clear();
+ defaultReadPos.clear();
+ }
+
public long getWritePoint() {
- return ((long)currentWriteBufferIndex << wbSizeLog2) + currentWriteOffset;
+ return ((long)writePos.bufferIndex << wbSizeLog2) + writePos.offset;
}
public long getReadPoint() {
- return ((long)currentReadBufferIndex << wbSizeLog2) + currentReadOffset;
+ return getReadPoint(defaultReadPos);
+ }
+
+ public long getReadPoint(Position readPos) {
+ return (readPos.bufferIndex * (long)wbSize) + readPos.offset;
}
public void writeVLong(long value) {
@@ -312,22 +342,22 @@ public final class WriteBuffers implemen
readBuffer = writeBuffers.get(readBufIndex);
readBufOffset = 0;
}
- if (currentWriteOffset == wbSize) {
+ if (writePos.offset == wbSize) {
nextBufferToWrite();
}
// How much we can read from current read buffer, out of what we need.
int toRead = Math.min(length - srcOffset, wbSize - readBufOffset);
// How much we can write to current write buffer, out of what we need.
- int toWrite = Math.min(toRead, wbSize - currentWriteOffset);
- System.arraycopy(readBuffer, readBufOffset, currentWriteBuffer, currentWriteOffset, toWrite);
- currentWriteOffset += toWrite;
+ int toWrite = Math.min(toRead, wbSize - writePos.offset);
+ System.arraycopy(readBuffer, readBufOffset, writePos.buffer, writePos.offset, toWrite);
+ writePos.offset += toWrite;
readBufOffset += toWrite;
srcOffset += toWrite;
if (toRead > toWrite) {
nextBufferToWrite();
toRead -= toWrite; // Remains to copy from current read buffer. Less than wbSize by def.
- System.arraycopy(readBuffer, readBufOffset, currentWriteBuffer, currentWriteOffset, toRead);
- currentWriteOffset += toRead;
+ System.arraycopy(readBuffer, readBufOffset, writePos.buffer, writePos.offset, toRead);
+ writePos.offset += toRead;
readBufOffset += toRead;
srcOffset += toRead;
}
@@ -404,58 +434,57 @@ public final class WriteBuffers implemen
}
}
- private boolean isAllInOneReadBuffer(int length) {
- return currentReadOffset + length <= wbSize;
+ private boolean isAllInOneReadBuffer(int length, Position readPos) {
+ return readPos.offset + length <= wbSize;
}
private boolean isAllInOneWriteBuffer(int length) {
- return currentWriteOffset + length <= wbSize;
+ return writePos.offset + length <= wbSize;
}
public void seal() {
- if (currentWriteOffset < (wbSize * 0.8)) { // arbitrary
- byte[] smallerBuffer = new byte[currentWriteOffset];
- System.arraycopy(currentWriteBuffer, 0, smallerBuffer, 0, currentWriteOffset);
- writeBuffers.set(currentWriteBufferIndex, smallerBuffer);
+ if (writePos.offset < (wbSize * 0.8)) { // arbitrary
+ byte[] smallerBuffer = new byte[writePos.offset];
+ System.arraycopy(writePos.buffer, 0, smallerBuffer, 0, writePos.offset);
+ writeBuffers.set(writePos.bufferIndex, smallerBuffer);
}
- if (currentWriteBufferIndex + 1 < writeBuffers.size()) {
- writeBuffers.subList(currentWriteBufferIndex + 1, writeBuffers.size()).clear();
+ if (writePos.bufferIndex + 1 < writeBuffers.size()) {
+ writeBuffers.subList(writePos.bufferIndex + 1, writeBuffers.size()).clear();
}
- currentWriteBuffer = currentReadBuffer = null; // Make sure we don't reference any old buffer.
- currentWriteBufferIndex = currentReadBufferIndex = -1;
- currentReadOffset = currentWriteOffset = -1;
+ // Make sure we don't reference any old buffer.
+ clearState();
}
- public long readFiveByteULong(long offset) {
- return readNByteLong(offset, 5);
+ public long readNByteLong(long offset, int bytes) {
+ return readNByteLong(offset, bytes, defaultReadPos);
}
- private long readNByteLong(long offset, int bytes) {
- setReadPoint(offset);
+ public long readNByteLong(long offset, int bytes, Position readPos) {
+ setReadPoint(offset, readPos);
long v = 0;
- if (isAllInOneReadBuffer(bytes)) {
+ if (isAllInOneReadBuffer(bytes, readPos)) {
for (int i = 0; i < bytes; ++i) {
- v = (v << 8) + (currentReadBuffer[currentReadOffset + i] & 0xff);
+ v = (v << 8) + (readPos.buffer[readPos.offset + i] & 0xff);
}
- currentReadOffset += bytes;
+ readPos.offset += bytes;
} else {
for (int i = 0; i < bytes; ++i) {
- v = (v << 8) + (readNextByte() & 0xff);
+ v = (v << 8) + (readNextByte(readPos) & 0xff);
}
}
return v;
}
public void writeFiveByteULong(long offset, long v) {
- int prevIndex = currentWriteBufferIndex, prevOffset = currentWriteOffset;
+ int prevIndex = writePos.bufferIndex, prevOffset = writePos.offset;
setWritePoint(offset);
if (isAllInOneWriteBuffer(5)) {
- currentWriteBuffer[currentWriteOffset] = (byte)(v >>> 32);
- currentWriteBuffer[currentWriteOffset + 1] = (byte)(v >>> 24);
- currentWriteBuffer[currentWriteOffset + 2] = (byte)(v >>> 16);
- currentWriteBuffer[currentWriteOffset + 3] = (byte)(v >>> 8);
- currentWriteBuffer[currentWriteOffset + 4] = (byte)(v);
- currentWriteOffset += 5;
+ writePos.buffer[writePos.offset] = (byte)(v >>> 32);
+ writePos.buffer[writePos.offset + 1] = (byte)(v >>> 24);
+ writePos.buffer[writePos.offset + 2] = (byte)(v >>> 16);
+ writePos.buffer[writePos.offset + 3] = (byte)(v >>> 8);
+ writePos.buffer[writePos.offset + 4] = (byte)(v);
+ writePos.offset += 5;
} else {
setByte(offset++, (byte)(v >>> 32));
setByte(offset++, (byte)(v >>> 24));
@@ -463,9 +492,9 @@ public final class WriteBuffers implemen
setByte(offset++, (byte)(v >>> 8));
setByte(offset, (byte)(v));
}
- currentWriteBufferIndex = prevIndex;
- currentWriteBuffer = writeBuffers.get(currentWriteBufferIndex);
- currentWriteOffset = prevOffset;
+ writePos.bufferIndex = prevIndex;
+ writePos.buffer = writeBuffers.get(writePos.bufferIndex);
+ writePos.offset = prevOffset;
}
public int readInt(long offset) {
@@ -474,23 +503,23 @@ public final class WriteBuffers implemen
@Override
public void writeInt(long offset, int v) {
- int prevIndex = currentWriteBufferIndex, prevOffset = currentWriteOffset;
+ int prevIndex = writePos.bufferIndex, prevOffset = writePos.offset;
setWritePoint(offset);
if (isAllInOneWriteBuffer(4)) {
- currentWriteBuffer[currentWriteOffset] = (byte)(v >> 24);
- currentWriteBuffer[currentWriteOffset + 1] = (byte)(v >> 16);
- currentWriteBuffer[currentWriteOffset + 2] = (byte)(v >> 8);
- currentWriteBuffer[currentWriteOffset + 3] = (byte)(v);
- currentWriteOffset += 4;
+ writePos.buffer[writePos.offset] = (byte)(v >> 24);
+ writePos.buffer[writePos.offset + 1] = (byte)(v >> 16);
+ writePos.buffer[writePos.offset + 2] = (byte)(v >> 8);
+ writePos.buffer[writePos.offset + 3] = (byte)(v);
+ writePos.offset += 4;
} else {
setByte(offset++, (byte)(v >>> 24));
setByte(offset++, (byte)(v >>> 16));
setByte(offset++, (byte)(v >>> 8));
setByte(offset, (byte)(v));
}
- currentWriteBufferIndex = prevIndex;
- currentWriteBuffer = writeBuffers.get(currentWriteBufferIndex);
- currentWriteOffset = prevOffset;
+ writePos.bufferIndex = prevIndex;
+ writePos.buffer = writeBuffers.get(writePos.bufferIndex);
+ writePos.offset = prevOffset;
}
// Lifted from org.apache.hadoop.util.hash.MurmurHash... but supports offset.
@@ -551,4 +580,8 @@ public final class WriteBuffers implemen
public long size() {
return writeBuffers.size() * (long) wbSize;
}
+
+ public Position getReadPosition() {
+ return defaultReadPos;
+ }
}
\ No newline at end of file