HIVE-20249 : LLAP IO: NPE during refCount decrement (Prasanth Jayachandran, reviewed by Sergey Shelukhin)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/042b2ef7 Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/042b2ef7 Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/042b2ef7 Branch: refs/heads/master Commit: 042b2ef7df6af8b93adeb936d94c4079153467ff Parents: d4b7b93 Author: sergey <[email protected]> Authored: Mon Jul 30 12:39:43 2018 -0700 Committer: sergey <[email protected]> Committed: Mon Jul 30 12:54:41 2018 -0700 ---------------------------------------------------------------------- .../hadoop/hive/llap/cache/BuddyAllocator.java | 26 +++++++++++--------- .../hive/llap/cache/LowLevelCacheImpl.java | 12 +++++++++ .../ql/io/orc/encoded/EncodedReaderImpl.java | 4 ++- 3 files changed, 29 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/042b2ef7/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java ---------------------------------------------------------------------- diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java index fcfc22a..013f353 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java @@ -1352,21 +1352,23 @@ public final class BuddyAllocator public void deallocate(LlapAllocatorBuffer buffer, boolean isAfterMove) { assert data != null; - int pos = buffer.byteBuffer.position(); - // Note: this is called by someone who has ensured the buffer is not going to be moved. - int headerIx = pos >>> minAllocLog2; - int freeListIx = freeListFromAllocSize(buffer.allocSize); - if (assertsEnabled && !isAfterMove) { - LlapAllocatorBuffer buf = buffers[headerIx]; - if (buf != buffer) { - failWithLog(arenaIx + ":" + headerIx + " => " + if (buffer != null && buffer.byteBuffer != null) { + int pos = buffer.byteBuffer.position(); + // Note: this is called by someone who has ensured the buffer is not going to be moved. + int headerIx = pos >>> minAllocLog2; + int freeListIx = freeListFromAllocSize(buffer.allocSize); + if (assertsEnabled && !isAfterMove) { + LlapAllocatorBuffer buf = buffers[headerIx]; + if (buf != buffer) { + failWithLog(arenaIx + ":" + headerIx + " => " + toDebugString(buffer) + ", " + toDebugString(buf)); + } + assertBufferLooksValid(freeListFromHeader(headers[headerIx]), buf, arenaIx, headerIx); + checkHeader(headerIx, freeListIx, true); } - assertBufferLooksValid(freeListFromHeader(headers[headerIx]), buf, arenaIx, headerIx); - checkHeader(headerIx, freeListIx, true); + buffers[headerIx] = null; + addToFreeListWithMerge(headerIx, freeListIx, buffer, CasLog.Src.DEALLOC); } - buffers[headerIx] = null; - addToFreeListWithMerge(headerIx, freeListIx, buffer, CasLog.Src.DEALLOC); } private void addToFreeListWithMerge(int headerIx, int freeListIx, http://git-wip-us.apache.org/repos/asf/hive/blob/042b2ef7/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java ---------------------------------------------------------------------- diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java index 53bdc2a..e012d7d 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java @@ -20,6 +20,7 @@ package org.apache.hadoop.hive.llap.cache; import org.apache.orc.impl.RecordReaderUtils; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -40,6 +41,7 @@ import org.apache.hive.common.util.Ref; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Function; +import com.google.common.base.Joiner; public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, LlapIoDebugDump { private static final int DEFAULT_CLEANUP_INTERVAL = 600; @@ -457,6 +459,10 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla try { int fileLocked = 0, fileUnlocked = 0, fileEvicted = 0, fileMoving = 0; if (e.getValue().getCache().isEmpty()) continue; + List<LlapDataBuffer> lockedBufs = null; + if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) { + lockedBufs = new ArrayList<>(); + } for (Map.Entry<Long, LlapDataBuffer> e2 : e.getValue().getCache().entrySet()) { int newRc = e2.getValue().tryIncRef(); if (newRc < 0) { @@ -470,6 +476,9 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla try { if (newRc > 1) { // We hold one refcount. ++fileLocked; + if (lockedBufs != null) { + lockedBufs.add(e2.getValue()); + } } else { ++fileUnlocked; } @@ -483,6 +492,9 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla allMoving += fileMoving; sb.append("\n file " + e.getKey() + ": " + fileLocked + " locked, " + fileUnlocked + " unlocked, " + fileEvicted + " evicted, " + fileMoving + " being moved"); + if (fileLocked > 0 && LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) { + LlapIoImpl.LOCKING_LOGGER.trace("locked-buffers: {}", lockedBufs); + } } finally { e.getValue().decRef(); } http://git-wip-us.apache.org/repos/asf/hive/blob/042b2ef7/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 348f9df..759594a 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 @@ -1964,7 +1964,9 @@ class EncodedReaderImpl implements EncodedReader { } finally { // Release the unreleased buffers. See class comment about refcounts. try { - releaseInitialRefcounts(toRead.next); + if (toRead != null) { + releaseInitialRefcounts(toRead.next); + } releaseBuffers(toRelease.keySet(), true); } catch (Throwable t) { if (!hasError) throw new IOException(t);
