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 2017/03/21 21:02:05 UTC
hive git commit: HIVE-16180 : LLAP: Native memory leak in
EncodedReader (Sergey Shelukhin/Prasanth Jayachandran,
reviewed by Prasanth Jayachandran/Sergey Shelukhin)
Repository: hive
Updated Branches:
refs/heads/master 4f45536f1 -> 9f5a3e3d8
HIVE-16180 : LLAP: Native memory leak in EncodedReader (Sergey Shelukhin/Prasanth Jayachandran, reviewed by Prasanth Jayachandran/Sergey Shelukhin)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/9f5a3e3d
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/9f5a3e3d
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/9f5a3e3d
Branch: refs/heads/master
Commit: 9f5a3e3d89db7d6f4754eb345ad9abb6997857e1
Parents: 4f45536
Author: Sergey Shelukhin <se...@apache.org>
Authored: Tue Mar 21 13:55:27 2017 -0700
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Tue Mar 21 13:55:27 2017 -0700
----------------------------------------------------------------------
.../hadoop/hive/llap/cache/SimpleAllocator.java | 5 +-
.../ql/io/orc/encoded/EncodedReaderImpl.java | 188 ++++++++++++-------
2 files changed, 126 insertions(+), 67 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/9f5a3e3d/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleAllocator.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleAllocator.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleAllocator.java
index d8f59d1..51eb34e 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleAllocator.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleAllocator.java
@@ -32,9 +32,10 @@ public final class SimpleAllocator implements Allocator, BuddyAllocatorMXBean {
private final boolean isDirect;
private static Field cleanerField;
static {
- ByteBuffer tmp = ByteBuffer.allocateDirect(1);
try {
- cleanerField = tmp.getClass().getDeclaredField("cleaner");
+ // TODO: To make it work for JDK9 use CleanerUtil from https://issues.apache.org/jira/browse/HADOOP-12760
+ final Class<?> dbClazz = Class.forName("java.nio.DirectByteBuffer");
+ cleanerField = dbClazz.getDeclaredField("cleaner");
cleanerField.setAccessible(true);
} catch (Throwable t) {
LlapIoImpl.LOG.warn("Cannot initialize DirectByteBuffer cleaner", t);
http://git-wip-us.apache.org/repos/asf/hive/blob/9f5a3e3d/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
index 0ac3ec5..19f3451 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
@@ -18,9 +18,12 @@
package org.apache.hadoop.hive.ql.io.orc.encoded;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.IdentityHashMap;
import java.util.List;
import org.slf4j.Logger;
@@ -43,11 +46,13 @@ import org.apache.orc.impl.RecordReaderUtils;
import org.apache.orc.impl.StreamName;
import org.apache.orc.StripeInformation;
import org.apache.orc.impl.BufferChunk;
-import org.apache.hadoop.hive.llap.DebugUtils;
import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.OrcEncodedColumnBatch;
import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.PoolFactory;
import org.apache.orc.OrcProto;
+import sun.misc.Cleaner;
+
+
/**
* Encoded reader implementation.
*
@@ -80,6 +85,17 @@ import org.apache.orc.OrcProto;
*/
class EncodedReaderImpl implements EncodedReader {
public static final Logger LOG = LoggerFactory.getLogger(EncodedReaderImpl.class);
+ private static Field cleanerField;
+ static {
+ try {
+ // TODO: To make it work for JDK9 use CleanerUtil from https://issues.apache.org/jira/browse/HADOOP-12760
+ final Class<?> dbClazz = Class.forName("java.nio.DirectByteBuffer");
+ cleanerField = dbClazz.getDeclaredField("cleaner");
+ cleanerField.setAccessible(true);
+ } catch (Throwable t) {
+ cleanerField = null;
+ }
+ }
private static final Object POOLS_CREATION_LOCK = new Object();
private static Pools POOLS;
private static class Pools {
@@ -303,15 +319,35 @@ class EncodedReaderImpl implements EncodedReader {
}
}
+ // TODO: the memory release could be optimized - we could release original buffers after we
+ // are fully done with each original buffer from disk. For now release all at the end;
+ // it doesn't increase the total amount of memory we hold, just the duration a bit.
+ // This is much simpler - we can just remember original ranges after reading them, and
+ // release them at the end. In a few cases where it's easy to determine that a buffer
+ // can be freed in advance, we remove it from the map.
+ IdentityHashMap<ByteBuffer, Boolean> toRelease = null;
if (!isAllInCache.value) {
if (!isDataReaderOpen) {
this.dataReader.open();
isDataReaderOpen = true;
}
dataReader.readFileData(toRead.next, stripeOffset, cacheWrapper.getAllocator().isDirectAlloc());
+ toRelease = new IdentityHashMap<>();
+ DiskRangeList drl = toRead.next;
+ while (drl != null) {
+ if (drl instanceof BufferChunk) {
+ toRelease.put(drl.getData(), true);
+ }
+ drl = drl.next;
+ }
}
// 3. For uncompressed case, we need some special processing before read.
+ // Basically, we are trying to create artificial, consistent ranges to cache, as there are
+ // no CBs in an uncompressed file. At the end of this processing, the list would contain
+ // either cache buffers, or buffers allocated by us and not cached (if we are only reading
+ // parts of the data for some ranges and don't want to cache it). Both are represented by
+ // CacheChunks, so the list is just CacheChunk-s from that point on.
DiskRangeList iter = toRead.next; // Keep "toRead" list for future use, don't extract().
if (codec == null) {
for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
@@ -326,6 +362,12 @@ class EncodedReaderImpl implements EncodedReader {
}
}
}
+ // Release buffers as we are done with all the streams... also see toRelease comment.\
+ // With uncompressed streams, we know we are done earlier.
+ if (toRelease != null) {
+ releaseBuffers(toRelease.keySet(), true);
+ toRelease = null;
+ }
if (isTracingEnabled) {
LOG.trace("Disk ranges after pre-read (file " + fileKey + ", base offset "
+ stripeOffset + "): " + RecordReaderUtils.stringifyDiskRanges(toRead.next));
@@ -383,7 +425,7 @@ class EncodedReaderImpl implements EncodedReader {
long unlockUntilCOffset = sctx.offset + sctx.length;
DiskRangeList lastCached = readEncodedStream(stripeOffset, iter,
sctx.offset, sctx.offset + sctx.length, sctx.stripeLevelStream,
- unlockUntilCOffset, sctx.offset);
+ unlockUntilCOffset, sctx.offset, toRelease);
if (lastCached != null) {
iter = lastCached;
}
@@ -411,7 +453,7 @@ class EncodedReaderImpl implements EncodedReader {
boolean isStartOfStream = sctx.bufferIter == null;
DiskRangeList lastCached = readEncodedStream(stripeOffset,
(isStartOfStream ? iter : sctx.bufferIter), cOffset, endCOffset, cb,
- unlockUntilCOffset, sctx.offset);
+ unlockUntilCOffset, sctx.offset, toRelease);
if (lastCached != null) {
sctx.bufferIter = iter = lastCached;
}
@@ -438,6 +480,10 @@ class EncodedReaderImpl implements EncodedReader {
// Release the unreleased buffers. See class comment about refcounts.
releaseInitialRefcounts(toRead.next);
+ // Release buffers as we are done with all the streams... also see toRelease comment.
+ if (toRelease != null) {
+ releaseBuffers(toRelease.keySet(), true);
+ }
releaseCacheChunksIntoObjectPool(toRead.next);
}
@@ -605,8 +651,8 @@ class EncodedReaderImpl implements EncodedReader {
* the master list, so they are safe to keep as iterators for various streams.
*/
public DiskRangeList readEncodedStream(long baseOffset, DiskRangeList start, long cOffset,
- long endCOffset, ColumnStreamData csd, long unlockUntilCOffset, long streamOffset)
- throws IOException {
+ long endCOffset, ColumnStreamData csd, long unlockUntilCOffset, long streamOffset,
+ IdentityHashMap<ByteBuffer, Boolean> toRelease) throws IOException {
if (csd.getCacheBuffers() == null) {
csd.setCacheBuffers(new ArrayList<MemoryBuffer>());
} else {
@@ -615,10 +661,10 @@ class EncodedReaderImpl implements EncodedReader {
if (cOffset == endCOffset) return null;
boolean isCompressed = codec != null;
List<ProcCacheChunk> toDecompress = null;
- List<ByteBuffer> toRelease = null;
List<IncompleteCb> badEstimates = null;
+ List<ByteBuffer> toReleaseCopies = null;
if (isCompressed) {
- toRelease = !dataReader.isTrackingDiskRanges() ? null : new ArrayList<ByteBuffer>();
+ toReleaseCopies = new ArrayList<>();
toDecompress = new ArrayList<>();
badEstimates = new ArrayList<>();
}
@@ -636,8 +682,8 @@ class EncodedReaderImpl implements EncodedReader {
// 2. Go thru the blocks; add stuff to results and prepare the decompression work (see below).
try {
lastUncompressed = isCompressed ?
- prepareRangesForCompressedRead(cOffset, endCOffset, streamOffset,
- unlockUntilCOffset, current, csd, toRelease, toDecompress, badEstimates)
+ prepareRangesForCompressedRead(cOffset, endCOffset, streamOffset, unlockUntilCOffset,
+ current, csd, toRelease, toReleaseCopies, toDecompress, badEstimates)
: prepareRangesForUncompressedRead(
cOffset, endCOffset, streamOffset, unlockUntilCOffset, current, csd);
} catch (Exception ex) {
@@ -657,7 +703,10 @@ class EncodedReaderImpl implements EncodedReader {
assert result == null; // We don't expect conflicts from bad estimates.
}
- if (toDecompress == null || toDecompress.isEmpty()) return lastUncompressed; // Nothing to do.
+ if (toDecompress == null || toDecompress.isEmpty()) {
+ releaseBuffers(toReleaseCopies, false);
+ return lastUncompressed; // Nothing to do.
+ }
// 3. Allocate the buffers, prepare cache keys.
// At this point, we have read all the CBs we need to read. cacheBuffers contains some cache
@@ -690,21 +739,18 @@ class EncodedReaderImpl implements EncodedReader {
cacheWrapper.reuseBuffer(chunk.getBuffer());
}
- // 5. Release original compressed buffers to zero-copy reader if needed.
- if (toRelease != null) {
- assert dataReader.isTrackingDiskRanges();
- for (ByteBuffer buffer : toRelease) {
- dataReader.releaseBuffer(buffer);
- }
- }
+ // 5. Release the copies we made directly to the cleaner.
+ releaseBuffers(toReleaseCopies, false);
// 6. Finally, put uncompressed data to cache.
if (fileKey != null) {
- long[] collisionMask = cacheWrapper.putFileData(fileKey, cacheKeys, targetBuffers, baseOffset);
+ long[] collisionMask = cacheWrapper.putFileData(
+ fileKey, cacheKeys, targetBuffers, baseOffset);
processCacheCollisions(collisionMask, toDecompress, targetBuffers, csd.getCacheBuffers());
}
- // 7. It may happen that we know we won't use some compression buffers anymore.
+ // 7. It may happen that we know we won't use some cache buffers anymore (the alternative
+ // is that we will use the same buffers for other streams in separate calls).
// Release initial refcounts.
for (ProcCacheChunk chunk : toDecompress) {
ponderReleaseInitialRefcount(unlockUntilCOffset, streamOffset, chunk);
@@ -713,9 +759,11 @@ class EncodedReaderImpl implements EncodedReader {
return lastUncompressed;
}
+ /** Subset of readEncodedStream specific to compressed streams, separate to avoid long methods. */
private CacheChunk prepareRangesForCompressedRead(long cOffset, long endCOffset,
- long streamOffset, long unlockUntilCOffset, DiskRangeList current, ColumnStreamData columnStreamData,
- List<ByteBuffer> toRelease, List<ProcCacheChunk> toDecompress,
+ long streamOffset, long unlockUntilCOffset, DiskRangeList current,
+ ColumnStreamData columnStreamData, IdentityHashMap<ByteBuffer, Boolean> toRelease,
+ List<ByteBuffer> toReleaseCopies, List<ProcCacheChunk> toDecompress,
List<IncompleteCb> badEstimates) throws IOException {
if (cOffset > current.getOffset()) {
// Target compression block is in the middle of the range; slice the range in two.
@@ -762,8 +810,8 @@ class EncodedReaderImpl implements EncodedReader {
throw new RuntimeException(msg);
}
BufferChunk bc = (BufferChunk)current;
- ProcCacheChunk newCached = addOneCompressionBuffer(
- bc, columnStreamData.getCacheBuffers(), toDecompress, toRelease, badEstimates);
+ ProcCacheChunk newCached = addOneCompressionBuffer(bc, columnStreamData.getCacheBuffers(),
+ toDecompress, toRelease, toReleaseCopies, badEstimates);
lastUncompressed = (newCached == null) ? lastUncompressed : newCached;
next = (newCached != null) ? newCached.next : null;
currentOffset = (next != null) ? next.getOffset() : -1;
@@ -777,9 +825,12 @@ class EncodedReaderImpl implements EncodedReader {
return lastUncompressed;
}
+ /** Subset of readEncodedStream specific to uncompressed streams, separate to avoid long methods. */
private CacheChunk prepareRangesForUncompressedRead(long cOffset, long endCOffset,
- long streamOffset, long unlockUntilCOffset, DiskRangeList current, ColumnStreamData columnStreamData)
- throws IOException {
+ long streamOffset, long unlockUntilCOffset, DiskRangeList current,
+ ColumnStreamData columnStreamData) throws IOException {
+ // Note: we are called after preReadUncompressedStream, so it doesn't have to do nearly as much
+ // as prepareRangesForCompressedRead does; e.g. every buffer is already a CacheChunk.
long currentOffset = cOffset;
CacheChunk lastUncompressed = null;
boolean isFirst = true;
@@ -819,11 +870,10 @@ class EncodedReaderImpl implements EncodedReader {
* We could avoid copy in non-zcr case and manage the buffer that was not allocated by our
* allocator. Uncompressed case is not mainline though so let's not complicate it.
*/
- private DiskRangeList preReadUncompressedStream(long baseOffset,
- DiskRangeList start, long streamOffset, long streamEnd) throws IOException {
+ private DiskRangeList preReadUncompressedStream(long baseOffset, DiskRangeList start,
+ long streamOffset, long streamEnd) throws IOException {
if (streamOffset == streamEnd) return null;
List<UncompressedCacheChunk> toCache = null;
- List<ByteBuffer> toRelease = null;
// 1. Find our bearings in the stream.
DiskRangeList current = findIntersectingPosition(start, streamOffset, streamEnd);
@@ -860,9 +910,6 @@ class EncodedReaderImpl implements EncodedReader {
if (current.getOffset() >= partEnd) {
continue; // We have no data at all for this part of the stream (could be unneeded), skip.
}
- if (toRelease == null && dataReader.isTrackingDiskRanges()) {
- toRelease = new ArrayList<ByteBuffer>();
- }
// We have some disk buffers... see if we have entire part, etc.
UncompressedCacheChunk candidateCached = null; // We will cache if we have the entire part.
DiskRangeList next = current;
@@ -877,21 +924,15 @@ class EncodedReaderImpl implements EncodedReader {
current = next;
if (noMoreDataForPart) break; // Done with this part.
- boolean wasSplit = false;
if (current.getEnd() > partEnd) {
// If the current buffer contains multiple parts, split it.
current = current.split(partEnd);
- wasSplit = true;
}
if (isTracingEnabled) {
LOG.trace("Processing uncompressed file data at ["
+ current.getOffset() + ", " + current.getEnd() + ")");
}
BufferChunk curBc = (BufferChunk)current;
- if (!wasSplit && toRelease != null) {
- toRelease.add(curBc.getChunk()); // TODO: is it valid to give zcr the modified 2nd part?
- }
-
// Track if we still have the entire part.
long hadEntirePartTo = hasEntirePartTo;
// We have data until the end of current block if we had it until the beginning.
@@ -952,15 +993,7 @@ class EncodedReaderImpl implements EncodedReader {
++ix;
}
- // 5. Release original compressed buffers to zero-copy reader if needed.
- if (toRelease != null) {
- assert dataReader.isTrackingDiskRanges();
- for (ByteBuffer buf : toRelease) {
- dataReader.releaseBuffer(buf);
- }
- }
-
- // 6. Finally, put uncompressed data to cache.
+ // 5. Put uncompressed data to cache.
if (fileKey != null) {
long[] collisionMask = cacheWrapper.putFileData(fileKey, cacheKeys, targetBuffers, baseOffset);
processCacheCollisions(collisionMask, toCache, targetBuffers, null);
@@ -969,7 +1002,6 @@ class EncodedReaderImpl implements EncodedReader {
return lastUncompressed;
}
-
private int determineUncompressedPartSize() {
// We will break the uncompressed data in the cache in the chunks that are the size
// of the prevalent ORC compression buffer (the default), or maximum allocation (since we
@@ -1178,7 +1210,8 @@ class EncodedReaderImpl implements EncodedReader {
*/
private ProcCacheChunk addOneCompressionBuffer(BufferChunk current,
List<MemoryBuffer> cacheBuffers, List<ProcCacheChunk> toDecompress,
- List<ByteBuffer> toRelease, List<IncompleteCb> badEstimates) throws IOException {
+ IdentityHashMap<ByteBuffer, Boolean> toRelease, List<ByteBuffer> toReleaseCopies,
+ List<IncompleteCb> badEstimates) throws IOException {
ByteBuffer slice = null;
ByteBuffer compressed = current.getChunk();
long cbStartOffset = current.getOffset();
@@ -1201,12 +1234,8 @@ class EncodedReaderImpl implements EncodedReader {
// Simple case - CB fits entirely in the disk range.
slice = compressed.slice();
slice.limit(chunkLength);
- ProcCacheChunk cc = addOneCompressionBlockByteBuffer(slice, isUncompressed,
+ return addOneCompressionBlockByteBuffer(slice, isUncompressed,
cbStartOffset, cbEndOffset, chunkLength, current, toDecompress, cacheBuffers);
- if (compressed.remaining() <= 0 && dataReader.isTrackingDiskRanges()) {
- toRelease.add(compressed);
- }
- return cc;
}
if (current.getEnd() < cbEndOffset && !current.hasContiguousNext()) {
badEstimates.add(addIncompleteCompressionBuffer(cbStartOffset, current, 0));
@@ -1216,6 +1245,7 @@ class EncodedReaderImpl implements EncodedReader {
// TODO: we could remove extra copy for isUncompressed case by copying directly to cache.
// We need to consolidate 2 or more buffers into one to decompress.
ByteBuffer copy = allocateBuffer(chunkLength, compressed.isDirect());
+ toReleaseCopies.add(copy); // We will always release copies at the end.
int remaining = chunkLength - compressed.remaining();
int originalPos = compressed.position();
copy.put(compressed);
@@ -1224,12 +1254,8 @@ class EncodedReaderImpl implements EncodedReader {
}
DiskRangeList next = current.next;
current.removeSelf();
- if (dataReader.isTrackingDiskRanges()) {
- if (originalPos == 0) {
- dataReader.releaseBuffer(compressed); // We copied the entire buffer.
- } else {
- toRelease.add(compressed); // There might be slices depending on this buffer.
- }
+ if (originalPos == 0 && toRelease.remove(compressed)) {
+ releaseBuffer(compressed, true);
}
int extraChunkCount = 0;
@@ -1246,15 +1272,15 @@ class EncodedReaderImpl implements EncodedReader {
copy.put(slice);
ProcCacheChunk cc = addOneCompressionBlockByteBuffer(copy, isUncompressed,
cbStartOffset, cbEndOffset, remaining, (BufferChunk)next, toDecompress, cacheBuffers);
- if (compressed.remaining() <= 0 && dataReader.isTrackingDiskRanges()) {
- dataReader.releaseBuffer(compressed); // We copied the entire buffer.
- }
+ if (compressed.remaining() <= 0 && toRelease.remove(compressed)) {
+ releaseBuffer(compressed, true); // We copied the entire buffer.
+ } // else there's more data to process; will be handled in next call.
return cc;
}
remaining -= compressed.remaining();
- copy.put(compressed);
- if (dataReader.isTrackingDiskRanges()) {
- dataReader.releaseBuffer(compressed); // We copied the entire buffer.
+ copy.put(compressed); // TODO: move into the if below; account for release call
+ if (toRelease.remove(compressed)) {
+ releaseBuffer(compressed, true); // We copied the entire buffer.
}
DiskRangeList tmp = next;
next = next.hasContiguousNext() ? next.next : null;
@@ -1270,6 +1296,38 @@ class EncodedReaderImpl implements EncodedReader {
}
}
+ private void releaseBuffers(Collection<ByteBuffer> toRelease, boolean isFromDataReader) {
+ if (toRelease == null) return;
+ for (ByteBuffer buf : toRelease) {
+ releaseBuffer(buf, isFromDataReader);
+ }
+ }
+
+ private void releaseBuffer(ByteBuffer bb, boolean isFromDataReader) {
+ if (isTracingEnabled) {
+ LOG.trace("Releasing the buffer " + System.identityHashCode(bb));
+ }
+ if (isFromDataReader && dataReader.isTrackingDiskRanges()) {
+ dataReader.releaseBuffer(bb);
+ return;
+ }
+ Field localCf = cleanerField;
+ if (!bb.isDirect() || localCf == null) return;
+ try {
+ Cleaner cleaner = (Cleaner) localCf.get(bb);
+ if (cleaner != null) {
+ cleaner.clean();
+ } else {
+ LOG.debug("Unable to clean a buffer using cleaner - no cleaner");
+ }
+ } catch (Exception e) {
+ // leave it for GC to clean up
+ LOG.warn("Unable to clean direct buffers using Cleaner.");
+ cleanerField = null;
+ }
+ }
+
+
private IncompleteCb addIncompleteCompressionBuffer(
long cbStartOffset, DiskRangeList target, int extraChunkCount) {
IncompleteCb icb = new IncompleteCb(cbStartOffset, target.getEnd());