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/02/28 00:08:34 UTC
svn commit: r1662854 - in /hive/branches/llap:
llap-server/src/java/org/apache/hadoop/hive/llap/cache/
llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/
llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/
ql/src/java/org/apache/hadoo...
Author: sershe
Date: Fri Feb 27 23:08:33 2015
New Revision: 1662854
URL: http://svn.apache.org/r1662854
Log:
Another attempt at fixing the decRef issue (forgot to replace the buffer in results)
Modified:
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/EncodedDataConsumer.java
hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataProducer.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java?rev=1662854&r1=1662853&r2=1662854&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java Fri Feb 27 23:08:33 2015
@@ -108,7 +108,8 @@ public final class LlapCacheableBuffer e
@Override
public String toString() {
- return "0x" + Integer.toHexString(System.identityHashCode(this));
+ int refCount = this.refCount.get();
+ return "0x" + Integer.toHexString(System.identityHashCode(this)) + "(" + refCount + ")";
}
/**
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/EncodedDataConsumer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/EncodedDataConsumer.java?rev=1662854&r1=1662853&r2=1662854&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/EncodedDataConsumer.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/EncodedDataConsumer.java Fri Feb 27 23:08:33 2015
@@ -101,7 +101,6 @@ public abstract class EncodedDataConsume
protected void returnProcessed(EncodedColumnBatch.StreamBuffer[][] data) {
for (EncodedColumnBatch.StreamBuffer[] sbs : data) {
for (EncodedColumnBatch.StreamBuffer sb : sbs) {
- if (sb.decRef() != 0) continue;
upstreamFeedback.returnData(sb);
}
}
@@ -152,9 +151,7 @@ public abstract class EncodedDataConsume
}
}
for (EncodedColumnBatch.StreamBuffer data : dataToDiscard) {
- if (data.decRef() == 0) {
- upstreamFeedback.returnData(data);
- }
+ upstreamFeedback.returnData(data);
}
}
Modified: hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataProducer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataProducer.java?rev=1662854&r1=1662853&r2=1662854&view=diff
==============================================================================
--- hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataProducer.java (original)
+++ hive/branches/llap/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataProducer.java Fri Feb 27 23:08:33 2015
@@ -414,6 +414,7 @@ public class OrcEncodedDataProducer impl
@Override
public void returnData(StreamBuffer data) {
+ if (data.decRef() != 0) return;
if (DebugUtils.isTraceLockingEnabled()) {
for (LlapMemoryBuffer buf : data.cacheBuffers) {
LlapIoImpl.LOG.info("Unlocking " + buf + " at the end of processing");
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java?rev=1662854&r1=1662853&r2=1662854&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/EncodedReaderImpl.java Fri Feb 27 23:08:33 2015
@@ -314,6 +314,10 @@ public class EncodedReaderImpl implement
consumer.consumeData(ecb);
}
}
+ if (DebugUtils.isTraceOrcEnabled()) {
+ LOG.info("Disk ranges after processing all the data "
+ + RecordReaderUtils.stringifyDiskRanges(toRead.next));
+ }
// TODO: this is not good; we hold all the blocks until we send them all.
// Hard to avoid due to sharing by RGs... perhaps we can still do better.
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java?rev=1662854&r1=1662853&r2=1662854&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java Fri Feb 27 23:08:33 2015
@@ -535,16 +535,24 @@ public abstract class InStream extends I
}
}
+ /**
+ * CacheChunk that is pre-created for new cache data; initially, it contains an original disk
+ * buffer and an unallocated LlapMemoryBuffer object. Before we expose it, the LMB is allocated,
+ * the data is decompressed, and original compressed data is discarded. The chunk lives on in
+ * the DiskRange list created for the request, and everyone treats it like regular CacheChunk.
+ */
private static class ProcCacheChunk extends CacheChunk {
public ProcCacheChunk(long cbStartOffset, long cbEndOffset, boolean isCompressed,
- ByteBuffer originalData, LlapMemoryBuffer targetBuffer) {
+ ByteBuffer originalData, LlapMemoryBuffer targetBuffer, int originalCbIndex) {
super(targetBuffer, cbStartOffset, cbEndOffset);
this.isCompressed = isCompressed;
this.originalData = originalData;
+ this.originalCbIndex = originalCbIndex;
}
boolean isCompressed;
ByteBuffer originalData = null;
+ int originalCbIndex;
}
/**
@@ -676,12 +684,14 @@ public abstract class InStream extends I
// 6. Finally, put data to cache.
long[] collisionMask = cache.putFileData(fileName, cacheKeys, targetBuffers, baseOffset);
- processCacheCollisions(cache, collisionMask, toDecompress, targetBuffers);
+ processCacheCollisions(
+ cache, collisionMask, toDecompress, targetBuffers, streamBuffer.cacheBuffers);
return lastCached;
}
private static void processCacheCollisions(LowLevelCache cache, long[] collisionMask,
- List<ProcCacheChunk> toDecompress, LlapMemoryBuffer[] targetBuffers) {
+ List<ProcCacheChunk> toDecompress, LlapMemoryBuffer[] targetBuffers,
+ List<LlapMemoryBuffer> cacheBuffers) {
if (collisionMask == null) return;
assert collisionMask.length >= (toDecompress.size() >>> 6);
// There are some elements that were cached in parallel, take care of them.
@@ -691,15 +701,25 @@ public abstract class InStream extends I
maskVal = collisionMask[i >>> 6];
}
if ((maskVal & 1) == 1) {
- // Cache has found an old buffer for the key and put it into array. Had the put succeeded
- // for our new buffer, it would have refcount of 2 - 1 from put, and 1 from notifyReused
- // call above. "Old" buffer now has refcount of 1 from put; new buffer is unchanged. We
- // will discard the new buffer, and lock old again to make it consistent.
+ // Cache has found an old buffer for the key and put it into array instead of our new one.
ProcCacheChunk replacedChunk = toDecompress.get(i);
LlapMemoryBuffer replacementBuffer = targetBuffers[i];
+ if (DebugUtils.isTraceOrcEnabled()) {
+ LOG.info("Discarding data due to cache collision: " + replacedChunk.buffer
+ + " replaced with " + replacementBuffer);
+ }
+ assert replacedChunk.buffer != replacementBuffer : i + " was not replaced in the results "
+ + "even though mask is [" + Long.toBinaryString(maskVal) + "]";
+ assert replacedChunk.originalCbIndex >= 0;
+ // Had the put succeeded for our new buffer, it would have refcount of 2 - 1 from put,
+ // and 1 from notifyReused call above. "Old" buffer now has the 1 from put; new buffer
+ // is unchanged at 1 from notifyReused. Make it all consistent.
cache.releaseBuffer(replacedChunk.buffer);
cache.notifyReused(replacementBuffer);
+ // Replace the buffer in our big range list, as well as in current results.
replacedChunk.buffer = replacementBuffer;
+ cacheBuffers.set(replacedChunk.originalCbIndex, replacementBuffer);
+ replacedChunk.originalCbIndex = -1; // This can only happen once at decompress time.
}
maskVal >>= 1;
}
@@ -845,8 +865,8 @@ public abstract class InStream extends I
// Add it to result in order we are processing.
cacheBuffers.add(futureAlloc);
// Add it to the list of work to decompress.
- ProcCacheChunk cc = new ProcCacheChunk(
- cbStartOffset, cbEndOffset, !isUncompressed, fullCompressionBlock, futureAlloc);
+ ProcCacheChunk cc = new ProcCacheChunk(cbStartOffset, cbEndOffset, !isUncompressed,
+ fullCompressionBlock, futureAlloc, cacheBuffers.size() - 1);
toDecompress.add(cc);
// Adjust the compression block position.
if (DebugUtils.isTraceOrcEnabled()) {
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java?rev=1662854&r1=1662853&r2=1662854&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderUtils.java Fri Feb 27 23:08:33 2015
@@ -260,22 +260,20 @@ public class RecordReaderUtils {
} else if (doForceDirect) {
ByteBuffer directBuf = ByteBuffer.allocateDirect(len);
// TODO: HDFS API is a mess, so handle all kinds of crap.
+ // Before 2.7, read() also doesn't adjust position correctly, so track it.
int pos = directBuf.position(), startPos = pos, endPos = pos + len;
try {
while (pos < endPos) {
int count = file.read(directBuf);
if (count < 0) throw new EOFException();
if (count == 0) {
- throw new AssertionError("HDFS won't read any bloody data "
- + (endPos - pos) + "@" + (pos - startPos) + " and " + pos);
- }
- if (directBuf.position() != pos) {
- RecordReaderImpl.LOG.info("Warning - position mismatch from " + file.getClass()
- + ": after reading " + count + ", expected " + pos + " but got " + directBuf.position());
+ throw new IOException(
+ "0-length read: " + (endPos - pos) + "@" + (pos - startPos) + " and " + pos);
}
pos += count;
if (pos > endPos) {
- throw new AssertionError("Position " + pos + " length " + len + "/" + endPos + " after reading " + count);
+ throw new AssertionError(
+ "Position " + pos + " length " + len + "/" + endPos + " after reading " + count);
}
directBuf.position(pos);
}