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 2018/08/24 18:32:06 UTC
[2/3] hive git commit: HIVE-16233 : llap: Query failed with
AllocatorOutOfMemoryException (Sergey Shelukhin,
reviewed by Prasanth Jayachandran and Gopal Vijayaraghavan)
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapAllocatorBuffer.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapAllocatorBuffer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapAllocatorBuffer.java
new file mode 100644
index 0000000..1490241
--- /dev/null
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapAllocatorBuffer.java
@@ -0,0 +1,396 @@
+package org.apache.hadoop.hive.llap.cache;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
+import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * We want to have cacheable and non-allocator buffers, as well as allocator buffers with no
+ * cache dependency, and also ones that are both. Alas, we could only achieve this if we were
+ * using a real programming language.
+ */
+public abstract class LlapAllocatorBuffer extends LlapCacheableBuffer implements MemoryBuffer {
+ private final AtomicLong state = new AtomicLong(0);
+
+ public ByteBuffer byteBuffer;
+ /** Allocator uses this to remember the allocation size. Somewhat redundant with header. */
+ public int allocSize;
+
+ public void initialize(ByteBuffer byteBuffer, int offset, int length) {
+ this.byteBuffer = byteBuffer.slice();
+ this.byteBuffer.position(offset);
+ this.byteBuffer.limit(offset + length);
+ this.allocSize = length;
+ }
+
+ public void initializeWithExistingSlice(ByteBuffer byteBuffer, int allocSize) {
+ this.byteBuffer = byteBuffer;
+ this.allocSize = allocSize;
+ }
+
+ public void setNewAllocLocation(int arenaIx, int headerIx) {
+ assert state.get() == 0 : "New buffer state is not 0 " + this;
+ long newState = State.setFlag(State.setLocation(0, arenaIx, headerIx), State.FLAG_NEW_ALLOC);
+ if (!state.compareAndSet(0, newState)) {
+ throw new AssertionError("Contention on the new buffer " + this);
+ }
+ }
+
+ @Override
+ public ByteBuffer getByteBufferDup() {
+ return byteBuffer.duplicate();
+ }
+
+ @Override
+ public ByteBuffer getByteBufferRaw() {
+ return byteBuffer;
+ }
+
+ @Override
+ public long getMemoryUsage() {
+ return allocSize;
+ }
+
+ public int incRef() {
+ return incRefInternal(true);
+ }
+
+ int tryIncRef() {
+ return incRefInternal(false);
+ }
+
+ static final int INCREF_EVICTED = -1, INCREF_FAILED = -2;
+ private int incRefInternal(boolean doWait) {
+ long newValue = -1;
+ while (true) {
+ long oldValue = state.get();
+ if (State.hasFlags(oldValue, State.FLAG_EVICTED)) return INCREF_EVICTED;
+ if (State.hasFlags(oldValue, State.FLAG_MOVING)) {
+ if (!doWait || !waitForState()) return INCREF_FAILED; // Thread is being interrupted.
+ continue;
+ }
+ int oldRefCount = State.getRefCount(oldValue);
+ assert oldRefCount >= 0 : "oldValue is " + oldValue + " " + this;
+ if (oldRefCount == State.MAX_REFCOUNT) throw new AssertionError(this);
+ newValue = State.incRefCount(oldValue);
+ if (State.hasFlags(oldValue, State.FLAG_NEW_ALLOC)) {
+ // Remove new-alloc flag on first use. Full unlock after that would imply force-discarding
+ // this buffer is acceptable. This is kind of an ugly compact between the cache and us.
+ newValue = State.switchFlag(newValue, State.FLAG_NEW_ALLOC);
+ }
+ if (state.compareAndSet(oldValue, newValue)) break;
+ }
+ int newRefCount = State.getRefCount(newValue);
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Locked {}; new ref count {}", this, newRefCount);
+ }
+ return newRefCount;
+ }
+
+ @VisibleForTesting
+ int getRefCount() {
+ return State.getRefCount(state.get());
+ }
+
+ @VisibleForTesting
+ @Override
+ public boolean isLocked() {
+ // Best-effort check. We cannot do a good check against caller thread, since
+ // refCount could still be > 0 if someone else locked. This is used for asserts and logs.
+ return State.getRefCount(state.get()) > 0;
+ }
+
+ @VisibleForTesting
+ public boolean isInvalid() {
+ return State.hasFlags(state.get(), State.FLAG_EVICTED);
+ }
+
+
+ public int decRef() {
+ long newState, oldState;
+ do {
+ oldState = state.get();
+ newState = State.decRefCount(oldState);
+ } while (!state.compareAndSet(oldState, newState));
+ int newRefCount = State.getRefCount(newState);
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Unlocked {}; refcount {}", this, newRefCount);
+ }
+ if (newRefCount < 0) {
+ throw new AssertionError("Unexpected refCount " + newRefCount + ": " + this);
+ }
+ return newRefCount;
+ }
+
+
+ /**
+ * Invalidates the cached buffer. The memory allocation in memory manager is managed by the
+ * caller, and therefore we mark the buffer as having released the memory too.
+ * @return Whether the we can invalidate; false if locked or already evicted.
+ */
+ @Override
+ public int invalidate() {
+ while (true) {
+ long oldValue = state.get();
+ if (State.getRefCount(oldValue) != 0) return INVALIDATE_FAILED;
+ if (State.hasFlags(oldValue, State.FLAG_EVICTED)) return INVALIDATE_ALREADY_INVALID;
+ long newValue = State.setFlag(oldValue, State.FLAG_EVICTED | State.FLAG_MEM_RELEASED);
+ if (state.compareAndSet(oldValue, newValue)) break;
+ }
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Invalidated {} due to eviction", this);
+ }
+ return INVALIDATE_OK;
+ }
+
+ /**
+ * Invalidates the uncached buffer. The memory allocation in memory manager is managed by the
+ * allocator; we will only mark it as released if there are no concurrent moves. In that case,
+ * the caller of this will release the memory; otherwise, the end of move will release.
+ * @return Arena index, if the buffer memory can be released; -1 otherwise.
+ */
+ public int invalidateAndRelease() {
+ boolean result;
+ long oldValue, newValue;
+ do {
+ result = false;
+ oldValue = state.get();
+ if (State.getRefCount(oldValue) != 0) {
+ throw new AssertionError("Refcount is " + State.getRefCount(oldValue));
+ }
+ if (State.hasFlags(oldValue, State.FLAG_EVICTED)) {
+ return -1; // Concurrent force-eviction - ignore.
+ }
+ newValue = State.setFlag(oldValue, State.FLAG_EVICTED);
+ if (!State.hasFlags(oldValue, State.FLAG_MOVING)) {
+ // No move pending, the allocator can release.
+ newValue = State.setFlag(newValue, State.FLAG_REMOVED | State.FLAG_MEM_RELEASED);
+ result = true;
+ }
+ } while (!state.compareAndSet(oldValue, newValue));
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Invalidated {} due to direct deallocation", this);
+ }
+ // Arena cannot change after we have marked it as released.
+ return result ? State.getArena(oldValue) : -1;
+ }
+
+ /**
+ * Marks previously invalidated buffer as released.
+ * @return Whether the buffer memory can be released.
+ */
+ public int releaseInvalidated() {
+ long oldValue, newValue;
+ do {
+ oldValue = state.get();
+ if (!State.hasFlags(oldValue, State.FLAG_EVICTED)) {
+ throw new AssertionError("Not invalidated");
+ }
+ if (State.hasFlags(oldValue, State.FLAG_MOVING | State.FLAG_REMOVED)) return -1;
+ // No move pending and no intervening discard, the allocator can release.
+ newValue = State.setFlag(oldValue, State.FLAG_REMOVED);
+ } while (!state.compareAndSet(oldValue, newValue));
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Removed {}", this);
+ }
+ // Arena cannot change after we have marked it as released.
+ return State.getArena(oldValue);
+ }
+
+ private boolean waitForState() {
+ synchronized (state) {
+ try {
+ state.wait(10);
+ return true;
+ } catch (InterruptedException e) {
+ LlapIoImpl.LOG.debug("Buffer incRef is deffering an interrupt");
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ long state = this.state.get();
+ int flags = State.getAllFlags(state);
+ return "0x" + Integer.toHexString(System.identityHashCode(this)) + "("
+ + State.getRefCount(state) + (flags == 0 ? "" : (", "
+ + State.toFlagString(flags))) + ")";
+ }
+
+
+ public String toDebugString() {
+ return toDebugString(state.get());
+ }
+
+ private String toDebugString(long state) {
+ return "0x" + Integer.toHexString(System.identityHashCode(this)) + "("
+ + State.getArena(state) + ":" + State.getHeader(state) + "; " + allocSize + "; "
+ + State.toFlagString(State.getAllFlags(state)) + ")";
+ }
+
+ public boolean startMoveOrDiscard(int arenaIx, int headerIx, boolean isForceDiscard) {
+ long oldValue, newValue;
+ do {
+ oldValue = state.get();
+ if (State.getRefCount(oldValue) != 0) {
+ return false;
+ }
+ int flags = State.getAllFlags(oldValue);
+ // The only allowed flag is new-alloc, and that only if we are not discarding.
+ if (flags != 0 && (isForceDiscard || flags != State.FLAG_NEW_ALLOC)) {
+ return false; // We could start a move if it's being evicted, but let's not do it for now.
+ }
+ if (State.getArena(oldValue) != arenaIx || State.getHeader(oldValue) != headerIx) {
+ return false; // The caller could re-check the location, but would probably find it locked.
+ }
+ newValue = State.setFlag(oldValue, State.FLAG_MOVING);
+ } while (!state.compareAndSet(oldValue, newValue));
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Locked {} in preparation for a move", this);
+ }
+ return true;
+ }
+
+ /**
+ * @return null if no action is required; otherwise, the buffer should be deallocated and,
+ * if the value is true, its memory should be released to the memory manager.
+ */
+ public Boolean cancelDiscard() {
+ long oldValue, newValue;
+ Boolean result;
+ do {
+ oldValue = state.get();
+ assert State.hasFlags(oldValue, State.FLAG_MOVING) : this.toDebugString();
+ newValue = State.switchFlag(oldValue, State.FLAG_MOVING);
+ result = null;
+ if (State.hasFlags(oldValue, State.FLAG_EVICTED)) {
+ if (State.hasFlags(oldValue, State.FLAG_REMOVED)) {
+ throw new AssertionError("Removed during the move " + this);
+ }
+ result = !State.hasFlags(oldValue, State.FLAG_MEM_RELEASED);
+ // Not necessary here cause noone will be looking at these after us; set them for clarity.
+ newValue = State.setFlag(newValue, State.FLAG_MEM_RELEASED | State.FLAG_REMOVED);
+ }
+ } while (!state.compareAndSet(oldValue, newValue));
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Move ended for {}", this);
+ }
+ synchronized (state) {
+ state.notifyAll();
+ }
+ return result;
+ }
+
+ /**
+ * @return null if no action is required; otherwise, the buffer should be deallocated and,
+ * if the value is true, its memory should be released to the memory manager.
+ */
+ public Boolean endDiscard() {
+ long oldValue, newValue;
+ Boolean result;
+ do {
+ oldValue = state.get();
+ assert State.hasFlags(oldValue, State.FLAG_MOVING);
+ newValue = State.switchFlag(oldValue, State.FLAG_MOVING);
+ newValue = State.setFlag(newValue,
+ State.FLAG_EVICTED | State.FLAG_MEM_RELEASED | State.FLAG_REMOVED);
+ result = null;
+ // See if someone else evicted this in parallel.
+ if (State.hasFlags(oldValue, State.FLAG_EVICTED)) {
+ if (State.hasFlags(oldValue, State.FLAG_REMOVED)) {
+ throw new AssertionError("Removed during the move " + this);
+ }
+ result = !State.hasFlags(oldValue, State.FLAG_MEM_RELEASED);
+ }
+ } while (!state.compareAndSet(oldValue, newValue));
+ if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
+ LlapIoImpl.LOCKING_LOGGER.trace("Discared {}", this);
+ }
+ synchronized (state) {
+ state.notifyAll();
+ }
+ return result;
+ }
+
+ private static final class State {
+ public static final int
+ FLAG_MOVING = 0b00001, // Locked by someone to move or force-evict.
+ FLAG_EVICTED = 0b00010, // Evicted. This is cache-specific.
+ FLAG_REMOVED = 0b00100, // Removed from allocator structures. The final state.
+ FLAG_MEM_RELEASED = 0b01000, // The memory was released to memory manager.
+ FLAG_NEW_ALLOC = 0b10000; // New allocation before the first use; cannot force-evict.
+ private static final int FLAGS_WIDTH = 5,
+ REFCOUNT_WIDTH = 19, ARENA_WIDTH = 16, HEADER_WIDTH = 24;
+
+ public static final long MAX_REFCOUNT = (1 << REFCOUNT_WIDTH) - 1;
+
+ private static final int REFCOUNT_SHIFT = FLAGS_WIDTH,
+ ARENA_SHIFT = REFCOUNT_SHIFT + REFCOUNT_WIDTH, HEADER_SHIFT = ARENA_SHIFT + ARENA_WIDTH;
+
+ private static final long FLAGS_MASK = (1L << FLAGS_WIDTH) - 1,
+ REFCOUNT_MASK = ((1L << REFCOUNT_WIDTH) - 1) << REFCOUNT_SHIFT,
+ ARENA_MASK = ((1L << ARENA_WIDTH) - 1) << ARENA_SHIFT,
+ HEADER_MASK = ((1L << HEADER_WIDTH) - 1) << HEADER_SHIFT;
+
+ public static boolean hasFlags(long value, int flags) {
+ return (value & flags) != 0;
+ }
+
+ public static int getAllFlags(long state) {
+ return (int) (state & FLAGS_MASK);
+ }
+ public static final int getRefCount(long state) {
+ return (int)((state & REFCOUNT_MASK) >>> REFCOUNT_SHIFT);
+ }
+ public static final int getArena(long state) {
+ return (int)((state & ARENA_MASK) >>> ARENA_SHIFT);
+ }
+ public static final int getHeader(long state) {
+ return (int)((state & HEADER_MASK) >>> HEADER_SHIFT);
+ }
+
+ public static final long incRefCount(long state) {
+ // Note: doesn't check for overflow. Could AND with max refcount mask but the caller checks.
+ return state + (1 << REFCOUNT_SHIFT);
+ }
+
+ public static final long decRefCount(long state) {
+ // Note: doesn't check for overflow. Could AND with max refcount mask but the caller checks.
+ return state - (1 << REFCOUNT_SHIFT);
+ }
+
+ public static final long setLocation(long state, int arenaIx, int headerIx) {
+ long arenaVal = ((long)arenaIx) << ARENA_SHIFT, arenaWMask = arenaVal & ARENA_MASK;
+ long headerVal = ((long)headerIx) << HEADER_SHIFT, headerWMask = headerVal & HEADER_MASK;
+ assert arenaVal == arenaWMask : "Arena " + arenaIx + " is wider than " + ARENA_WIDTH;
+ assert headerVal == headerWMask : "Header " + headerIx + " is wider than " + HEADER_WIDTH;
+ return (state & ~(ARENA_MASK | HEADER_MASK)) | arenaWMask | headerWMask;
+ }
+
+ public static final long setFlag(long state, int flags) {
+ assert flags <= FLAGS_MASK;
+ return state | flags;
+ }
+
+ public static final long switchFlag(long state, int flags) {
+ assert flags <= FLAGS_MASK;
+ return state ^ flags;
+ }
+
+ public static String toFlagString(int state) {
+ return StringUtils.leftPad(Integer.toBinaryString(state), REFCOUNT_SHIFT, '0');
+ }
+ }
+
+ @VisibleForTesting
+ int getArenaIndex() {
+ return State.getArena(state.get());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java
index 5c0b6f3..b53d4df 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapCacheableBuffer.java
@@ -19,6 +19,9 @@ package org.apache.hadoop.hive.llap.cache;
/**
* Buffer that can be managed by LowLevelEvictionPolicy.
+ * We want to have cacheable and non-allocator buffers, as well as allocator buffers with no
+ * cache dependency, and also ones that are both. Alas, we could only achieve this if we were
+ * using a real programming language.
*/
public abstract class LlapCacheableBuffer {
protected static final int IN_LIST = -2, NOT_IN_CACHE = -1;
@@ -38,7 +41,8 @@ public abstract class LlapCacheableBuffer {
/** Index in heap for LRFU/LFU cache policies. */
public int indexInHeap = NOT_IN_CACHE;
- protected abstract boolean invalidate();
+ public static final int INVALIDATE_OK = 0, INVALIDATE_FAILED = 1, INVALIDATE_ALREADY_INVALID = 2;
+ protected abstract int invalidate();
public abstract long getMemoryUsage();
public abstract void notifyEvicted(EvictionDispatcher evictionDispatcher);
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapDataBuffer.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapDataBuffer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapDataBuffer.java
index 7d5c101..0c232c8 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapDataBuffer.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapDataBuffer.java
@@ -18,132 +18,15 @@
package org.apache.hadoop.hive.llap.cache;
-import java.nio.ByteBuffer;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
-import org.apache.hadoop.hive.llap.DebugUtils;
-import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
-
-import com.google.common.annotations.VisibleForTesting;
-
-public final class LlapDataBuffer extends LlapCacheableBuffer implements MemoryBuffer {
-
- // For now, we don't track refcount for metadata blocks, don't clear them, don't reuse them and
- // basically rely on GC to remove them. So, refcount only applies to data blocks. If that
- // changes, refcount management should be move to LlapCacheableBuffer to be shared.
- private static final int EVICTED_REFCOUNT = -1;
+public final class LlapDataBuffer extends LlapAllocatorBuffer {
public static final int UNKNOWN_CACHED_LENGTH = -1;
- protected final AtomicInteger refCount = new AtomicInteger(0);
- public ByteBuffer byteBuffer;
- /** Allocator uses this to remember which arena to alloc from. */
- public int arenaIndex = -1;
- /** Allocator uses this to remember the allocation size. */
- public int allocSize;
/** ORC cache uses this to store compressed length; buffer is cached uncompressed, but
* the lookup is on compressed ranges, so we need to know this. */
public int declaredCachedLength = UNKNOWN_CACHED_LENGTH;
- public void initialize(
- int arenaIndex, ByteBuffer byteBuffer, int offset, int length) {
- this.byteBuffer = byteBuffer.slice();
- this.byteBuffer.position(offset);
- this.byteBuffer.limit(offset + length);
- this.arenaIndex = arenaIndex;
- this.allocSize = length;
- }
-
- @Override
- public ByteBuffer getByteBufferDup() {
- return byteBuffer.duplicate();
- }
-
- @Override
- public ByteBuffer getByteBufferRaw() {
- return byteBuffer;
- }
-
- @Override
- public long getMemoryUsage() {
- return allocSize;
- }
-
@Override
public void notifyEvicted(EvictionDispatcher evictionDispatcher) {
evictionDispatcher.notifyEvicted(this);
}
-
- int incRef() {
- int newRefCount = -1;
- while (true) {
- int oldRefCount = refCount.get();
- if (oldRefCount == EVICTED_REFCOUNT) return -1;
- assert oldRefCount >= 0 : "oldRefCount is " + oldRefCount + " " + this;
- newRefCount = oldRefCount + 1;
- if (refCount.compareAndSet(oldRefCount, newRefCount)) break;
- }
- if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
- LlapIoImpl.LOCKING_LOGGER.trace("Locked {}; new ref count {}", this, newRefCount);
- }
- return newRefCount;
- }
-
- @VisibleForTesting
- int getRefCount() {
- return refCount.get();
- }
-
- @VisibleForTesting
- @Override
- public boolean isLocked() {
- // Best-effort check. We cannot do a good check against caller thread, since
- // refCount could still be > 0 if someone else locked. This is used for asserts and logs.
- return refCount.get() > 0;
- }
-
- @VisibleForTesting
- public boolean isInvalid() {
- return refCount.get() == EVICTED_REFCOUNT;
- }
-
- int decRef() {
- int newRefCount = refCount.decrementAndGet();
- if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
- LlapIoImpl.LOCKING_LOGGER.trace("Unlocked {}; refcount {}", this, newRefCount);
- }
- if (newRefCount < 0) {
- throw new AssertionError("Unexpected refCount " + newRefCount + ": " + this);
- }
- return newRefCount;
- }
-
- /**
- * @return Whether the we can invalidate; false if locked or already evicted.
- */
- @Override
- public boolean invalidate() {
- while (true) {
- int value = refCount.get();
- if (value != 0) return false;
- if (refCount.compareAndSet(value, EVICTED_REFCOUNT)) break;
- }
- if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
- LlapIoImpl.LOCKING_LOGGER.trace("Invalidated {} due to eviction", this);
- }
- return true;
- }
-
- @Override
- public String toString() {
- int refCount = this.refCount.get();
- return "0x" + Integer.toHexString(System.identityHashCode(this)) + "(" + refCount + ")";
- }
-
- public static String toDataString(MemoryBuffer s) {
- if (s == null || s.getByteBufferRaw().remaining() == 0) return "" + s;
- byte b = s.getByteBufferRaw().get(s.getByteBufferRaw().position());
- int i = (b < 0) ? -b : b;
- return s + " (0x" + Integer.toHexString(i) + ")";
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/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 7673b5a..cc96d17 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
@@ -390,7 +390,7 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
private static final ByteBuffer fakeBuf = ByteBuffer.wrap(new byte[1]);
public static LlapDataBuffer allocateFake() {
LlapDataBuffer fake = new LlapDataBuffer();
- fake.initialize(-1, fakeBuf, 0, 1);
+ fake.initialize(fakeBuf, 0, 1);
return fake;
}
@@ -420,7 +420,6 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
throws InterruptedException {
// Iterate thru the file cache. This is best-effort.
Iterator<Map.Entry<Long, LlapDataBuffer>> subIter = fc.getCache().entrySet().iterator();
- boolean isEmpty = true;
while (subIter.hasNext()) {
long time = -1;
isPastEndTime.value = isPastEndTime.value || ((time = System.nanoTime()) >= endTime);
@@ -428,8 +427,6 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
? 1 : (endTime - time) / (1000000L * leftToCheck));
if (subIter.next().getValue().isInvalid()) {
subIter.remove();
- } else {
- isEmpty = false;
}
--leftToCheck;
}
@@ -478,17 +475,21 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
@Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nORC cache state ");
- int allLocked = 0, allUnlocked = 0, allEvicted = 0;
+ int allLocked = 0, allUnlocked = 0, allEvicted = 0, allMoving = 0;
for (Map.Entry<Object, FileCache<ConcurrentSkipListMap<Long, LlapDataBuffer>>> e :
cache.entrySet()) {
if (!e.getValue().incRef()) continue;
try {
- int fileLocked = 0, fileUnlocked = 0, fileEvicted = 0;
+ int fileLocked = 0, fileUnlocked = 0, fileEvicted = 0, fileMoving = 0;
if (e.getValue().getCache().isEmpty()) continue;
for (Map.Entry<Long, LlapDataBuffer> e2 : e.getValue().getCache().entrySet()) {
- int newRc = e2.getValue().incRef();
+ int newRc = e2.getValue().tryIncRef();
if (newRc < 0) {
- ++fileEvicted;
+ if (newRc == LlapAllocatorBuffer.INCREF_EVICTED) {
+ ++fileEvicted;
+ } else if (newRc == LlapAllocatorBuffer.INCREF_FAILED) {
+ ++fileMoving;
+ }
continue;
}
try {
@@ -504,13 +505,14 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
allLocked += fileLocked;
allUnlocked += fileUnlocked;
allEvicted += fileEvicted;
- sb.append("\n file " + e.getKey() + ": " + fileLocked + " locked, "
- + fileUnlocked + " unlocked, " + fileEvicted + " evicted");
+ allMoving += fileMoving;
+ sb.append("\n file " + e.getKey() + ": " + fileLocked + " locked, " + fileUnlocked
+ + " unlocked, " + fileEvicted + " evicted, " + fileMoving + " being moved");
} finally {
e.getValue().decRef();
}
}
- sb.append("\nORC cache summary: " + allLocked + " locked, "
- + allUnlocked + " unlocked, " + allEvicted + " evicted");
+ sb.append("\nORC cache summary: " + allLocked + " locked, " + allUnlocked + " unlocked, "
+ + allEvicted + " evicted, " + allMoving + " being moved");
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
index e30acb0..2c1086b 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
@@ -99,17 +99,15 @@ public class LowLevelCacheMemoryManager implements MemoryManager {
result = false;
break; // Test code path where we don't do more than one attempt.
}
- didDumpIoState = logEvictionIssue(++badCallCount, didDumpIoState);
- waitTimeMs = Math.min(1000, waitTimeMs << 1);
- assert waitTimeMs > 0;
- try {
- Thread.sleep(waitTimeMs);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+
+ if (isStopped != null && isStopped.get()) {
result = false;
break;
}
- if (isStopped != null && isStopped.get()) {
+ try {
+ Thread.sleep(badCallCount > 9 ? 1000 : (1 << badCallCount));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
result = false;
break;
}
@@ -139,46 +137,6 @@ public class LowLevelCacheMemoryManager implements MemoryManager {
return result;
}
-
- private boolean logEvictionIssue(int badCallCount, boolean didDumpIoState) {
- if (badCallCount <= LOCKING_DEBUG_DUMP_THRESHOLD) return didDumpIoState;
- String ioStateDump = maybeDumpIoState(didDumpIoState);
- if (ioStateDump == null) {
- LlapIoImpl.LOG.warn("Cannot evict blocks for " + badCallCount + " calls; cache full?");
- return didDumpIoState;
- } else {
- LlapIoImpl.LOG.warn("Cannot evict blocks; IO state:\n " + ioStateDump);
- return true;
- }
- }
-
- private String maybeDumpIoState(boolean didDumpIoState) {
- if (didDumpIoState) return null; // No more than once per reader.
- long now = System.nanoTime(), last = lastCacheDumpNs.get();
- while (true) {
- if (last != 0 && (now - last) < LOCKING_DEBUG_DUMP_PERIOD_NS) {
- return null; // We have recently dumped IO state into log.
- }
- if (lastCacheDumpNs.compareAndSet(last, now)) break;
- now = System.nanoTime();
- last = lastCacheDumpNs.get();
- }
- try {
- StringBuilder sb = new StringBuilder();
- memoryDumpRoot.debugDumpShort(sb);
- return sb.toString();
- } catch (Throwable t) {
- return "Failed to dump cache state: " + t.getClass() + " " + t.getMessage();
- }
- }
-
-
- @Override
- public long forceReservedMemory(int allocationSize, int count) {
- if (evictor == null) return 0;
- return evictor.tryEvictContiguousData(allocationSize, count);
- }
-
@Override
public void releaseMemory(final long memoryToRelease) {
long oldV;
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
index fd9d942..acbaf85 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
@@ -27,6 +27,4 @@ public interface LowLevelCachePolicy extends LlapOomDebugDump {
long evictSomeBlocks(long memoryToReserve);
void setEvictionListener(EvictionListener listener);
void setParentDebugDumper(LlapOomDebugDump dumper);
- /** TODO: temporary method until we have a better allocator */
- long tryEvictContiguousData(int allocationSize, int count);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
index 761fd00..e777dc7 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
@@ -24,7 +24,6 @@ import java.util.LinkedList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
@@ -83,8 +82,9 @@ public class LowLevelFifoCachePolicy implements LowLevelCachePolicy {
while (evicted < memoryToReserve && iter.hasNext()) {
LlapCacheableBuffer buffer = iter.next();
long memUsage = buffer.getMemoryUsage();
- if (memUsage < minSize || (minSize > 0 && !(buffer instanceof LlapDataBuffer))) continue;
- if (buffer.invalidate()) {
+ if (memUsage < minSize || (minSize > 0
+ && !(buffer instanceof LlapAllocatorBuffer))) continue;
+ if (LlapCacheableBuffer.INVALIDATE_OK == buffer.invalidate()) {
iter.remove();
evicted += memUsage;
evictionListener.notifyEvicted(buffer);
@@ -128,14 +128,4 @@ public class LowLevelFifoCachePolicy implements LowLevelCachePolicy {
parentDebugDump.debugDumpShort(sb);
}
}
-
- @Override
- public long tryEvictContiguousData(int allocationSize, int count) {
- long evicted = evictInternal(allocationSize * count, allocationSize);
- int remainingCount = count - (int)(evicted / allocationSize);
- if (remainingCount > 0) {
- evicted += evictInternal(allocationSize * remainingCount, -1);
- }
- return evicted;
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
index 3973c8a..158347a 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
@@ -18,17 +18,11 @@
package org.apache.hadoop.hive.llap.cache;
-import com.google.common.annotations.VisibleForTesting;
-
-import java.util.ArrayList;
-import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.DebugUtils;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
@@ -203,17 +197,6 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
return evicted;
}
- @Override
- public long tryEvictContiguousData(int allocationSize, int count) {
- int evicted = evictDataFromList(allocationSize, count);
- if (count <= evicted) return evicted * allocationSize;
- evicted += evictDataFromHeap(timer.get(), count - evicted, allocationSize);
- long evictedBytes = evicted * allocationSize;
- if (count <= evicted) return evictedBytes;
- evictedBytes += evictSomeBlocks(allocationSize * (count - evicted));
- return evictedBytes;
- }
-
private long evictFromList(long memoryToReserve) {
long evicted = 0;
LlapCacheableBuffer nextCandidate = null, firstCandidate = null;
@@ -224,8 +207,9 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
try {
nextCandidate = firstCandidate = listTail;
while (evicted < memoryToReserve && nextCandidate != null) {
- if (!nextCandidate.invalidate()) {
- // Locked buffer was in the list - just drop it; will be re-added on unlock.
+ if (LlapCacheableBuffer.INVALIDATE_OK != nextCandidate.invalidate()) {
+ // Locked, or invalidated, buffer was in the list - just drop it;
+ // will be re-added on unlock.
LlapCacheableBuffer lockedBuffer = nextCandidate;
if (firstCandidate == nextCandidate) {
firstCandidate = nextCandidate.prev;
@@ -258,42 +242,6 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
return evicted;
}
-
- private int evictDataFromList(int minSize, int count) {
- int evictedCount = 0;
- // Unlike the normal evictFromList, we don't necessarily evict a sequence of blocks. We won't
- // bother with optimization here and will just evict blocks one by one.
- List<LlapCacheableBuffer> evictedBuffers = new ArrayList<>(count);
- listLock.lock();
- try {
- LlapCacheableBuffer candidate = listTail;
- while (evictedCount < count && candidate != null) {
- LlapCacheableBuffer current = candidate;
- candidate = candidate.prev;
- long memUsage = current.getMemoryUsage();
- if (memUsage < minSize || !(current instanceof LlapDataBuffer)) continue;
- if (!current.invalidate()) {
- // Locked buffer was in the list - just drop it; will be re-added on unlock.
- removeFromListUnderLock(current);
- continue;
- }
- // Remove the buffer from the list.
- removeFromListUnderLock(current);
- // This makes granularity assumptions.
- assert memUsage % minSize == 0;
- evictedCount += (memUsage / minSize);
- evictedBuffers.add(current);
- }
- } finally {
- listLock.unlock();
- }
- for (LlapCacheableBuffer buffer : evictedBuffers) {
- evictionListener.notifyEvicted(buffer);
- }
- return evictedCount;
- }
-
-
// Note: rarely called (unless buffers are very large or we evict a lot, or in LFU case).
private LlapCacheableBuffer evictFromHeapUnderLock(long time) {
while (true) {
@@ -303,38 +251,6 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
}
}
- // Note: almost never called (unless buffers are very large or we evict a lot, or LFU).
- private int evictDataFromHeap(long time, int count, int minSize) {
- LlapCacheableBuffer evicted = null;
- int evictedCount = 0;
- synchronized (heap) {
- // Priorities go out of the window here.
- int index = 0;
- while (index < heapSize && evictedCount < count) {
- LlapCacheableBuffer buffer = heap[index];
- long memUsage = buffer.getMemoryUsage();
- if (memUsage < minSize || !(buffer instanceof LlapDataBuffer)) {
- ++index;
- continue;
- }
- LlapCacheableBuffer result = evictHeapElementUnderLock(time, index);
- // Don't advance the index - the buffer has been removed either way.
- if (result != null) {
- assert memUsage % minSize == 0;
- evictedCount += (memUsage / minSize);
- if (evicted != null) {
- evictionListener.notifyEvicted(evicted);
- }
- evicted = result;
- }
- }
- }
- if (evicted != null) {
- evictionListener.notifyEvicted(evicted);
- }
- return evictedCount;
- }
-
private void heapifyUpUnderLock(LlapCacheableBuffer buffer, long time) {
// See heapifyDown comment.
int ix = buffer.indexInHeap;
@@ -360,7 +276,8 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
}
result.indexInHeap = LlapCacheableBuffer.NOT_IN_CACHE;
--heapSize;
- boolean canEvict = result.invalidate();
+ int invalidateResult = result.invalidate();
+ boolean canEvict = invalidateResult == LlapCacheableBuffer.INVALIDATE_OK;
if (heapSize > 0) {
LlapCacheableBuffer newRoot = heap[heapSize];
newRoot.indexInHeap = ix;
@@ -370,7 +287,7 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
}
heapifyDownUnderLock(newRoot, time);
}
- // Otherwise we just removed a locked item from heap; unlock will re-add it, we continue.
+ // Otherwise we just removed a locked/invalid item from heap; we continue.
return canEvict ? result : null;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
index e1133cd..ff1f269 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
@@ -23,7 +23,5 @@ import java.util.concurrent.atomic.AtomicBoolean;
public interface MemoryManager extends LlapOomDebugDump {
void releaseMemory(long memUsage);
void updateMaxSize(long maxSize);
- /** TODO: temporary method until we get a better allocator. */
- long forceReservedMemory(int allocationSize, int count);
void reserveMemory(long memoryToReserve, AtomicBoolean isStopped);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
index cd5bc9b..7751845 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
@@ -18,11 +18,11 @@
package org.apache.hadoop.hive.llap.cache;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@@ -43,7 +43,7 @@ import org.apache.orc.OrcProto.ColumnEncoding;
import com.google.common.base.Function;
-public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
+public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapOomDebugDump {
private static final int DEFAULT_CLEANUP_INTERVAL = 600;
private final Allocator allocator;
private final AtomicInteger newEvictions = new AtomicInteger(0);
@@ -53,6 +53,14 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
private final long cleanupInterval;
private final LlapDaemonCacheMetrics metrics;
+ public static final class LlapSerDeDataBuffer extends LlapAllocatorBuffer {
+ public boolean isCached = false;
+ @Override
+ public void notifyEvicted(EvictionDispatcher evictionDispatcher) {
+ evictionDispatcher.notifyEvicted(this);
+ }
+ }
+
private static final class StripeInfoComparator implements
Comparator<StripeData> {
@Override
@@ -129,7 +137,7 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
private final long rowCount;
private final OrcProto.ColumnEncoding[] encodings;
- private LlapDataBuffer[][][] data; // column index, stream type, buffers
+ private LlapSerDeDataBuffer[][][] data; // column index, stream type, buffers
public StripeData(long knownTornStart, long firstStart, long lastStart, long lastEnd,
long rowCount, ColumnEncoding[] encodings) {
@@ -139,7 +147,7 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
this.lastEnd = lastEnd;
this.encodings = encodings;
this.rowCount = rowCount;
- this.data = encodings == null ? null : new LlapDataBuffer[encodings.length][][];
+ this.data = encodings == null ? null : new LlapSerDeDataBuffer[encodings.length][][];
}
@Override
@@ -172,7 +180,7 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
return encodings;
}
- public LlapDataBuffer[][][] getData() {
+ public LlapSerDeDataBuffer[][][] getData() {
return data;
}
@@ -191,18 +199,18 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
}
}
- public static String toString(LlapDataBuffer[][][] data) {
+ public static String toString(LlapSerDeDataBuffer[][][] data) {
if (data == null) return "null";
StringBuilder sb = new StringBuilder("[");
for (int i = 0; i < data.length; ++i) {
- LlapDataBuffer[][] colData = data[i];
+ LlapSerDeDataBuffer[][] colData = data[i];
if (colData == null) {
sb.append("null, ");
continue;
}
sb.append("colData [");
for (int j = 0; j < colData.length; ++j) {
- LlapDataBuffer[] streamData = colData[j];
+ LlapSerDeDataBuffer[] streamData = colData[j];
if (streamData == null) {
sb.append("null, ");
continue;
@@ -220,19 +228,18 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
}
- public static String toString(LlapDataBuffer[][] data) {
+ public static String toString(LlapSerDeDataBuffer[][] data) {
if (data == null) return "null";
StringBuilder sb = new StringBuilder("[");
for (int j = 0; j < data.length; ++j) {
- LlapDataBuffer[] streamData = data[j];
+ LlapSerDeDataBuffer[] streamData = data[j];
if (streamData == null) {
sb.append("null, ");
continue;
}
sb.append("[");
for (int k = 0; k < streamData.length; ++k) {
- LlapDataBuffer s = streamData[k];
- sb.append(LlapDataBuffer.toDataString(s));
+ sb.append(streamData[k]);
}
sb.append("], ");
}
@@ -405,11 +412,11 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
continue;
}
stripe.encodings[colIx] = cStripe.encodings[colIx];
- LlapDataBuffer[][] cColData = cStripe.data[colIx];
+ LlapSerDeDataBuffer[][] cColData = cStripe.data[colIx];
assert cColData != null;
for (int streamIx = 0;
cColData != null && streamIx < cColData.length; ++streamIx) {
- LlapDataBuffer[] streamData = cColData[streamIx];
+ LlapSerDeDataBuffer[] streamData = cColData[streamIx];
// Note: this relies on the fact that we always evict the entire column, so if
// we have the column data, we assume we have all the streams we need.
if (streamData == null) continue;
@@ -471,7 +478,7 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
}
}
- private boolean lockBuffer(LlapDataBuffer buffer, boolean doNotifyPolicy) {
+ private boolean lockBuffer(LlapSerDeDataBuffer buffer, boolean doNotifyPolicy) {
int rc = buffer.incRef();
if (rc > 0) {
metrics.incrCacheNumLockedBuffers();
@@ -561,28 +568,28 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
private void lockAllBuffersForPut(StripeData si, Priority priority) {
for (int i = 0; i < si.data.length; ++i) {
- LlapDataBuffer[][] colData = si.data[i];
+ LlapSerDeDataBuffer[][] colData = si.data[i];
if (colData == null) continue;
for (int j = 0; j < colData.length; ++j) {
- LlapDataBuffer[] streamData = colData[j];
+ LlapSerDeDataBuffer[] streamData = colData[j];
if (streamData == null) continue;
for (int k = 0; k < streamData.length; ++k) {
boolean canLock = lockBuffer(streamData[k], false); // false - not in cache yet
assert canLock;
cachePolicy.cache(streamData[k], priority);
- streamData[k].declaredCachedLength = streamData[k].getByteBufferRaw().remaining();
+ streamData[k].isCached = true;
}
}
}
}
private void handleRemovedStripeInfo(StripeData removed) {
- for (LlapDataBuffer[][] colData : removed.data) {
+ for (LlapSerDeDataBuffer[][] colData : removed.data) {
handleRemovedColumnData(colData);
}
}
- private void handleRemovedColumnData(LlapDataBuffer[][] removed) {
+ private void handleRemovedColumnData(LlapSerDeDataBuffer[][] removed) {
// TODO: could we tell the policy that we don't care about these and have them evicted? or we
// could just deallocate them when unlocked, and free memory + handle that in eviction.
// For now, just abandon the blocks - eventually, they'll get evicted.
@@ -603,7 +610,7 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
&& !to.encodings[colIx].equals(from.encodings[colIx])) {
throw new RuntimeException("Different encodings at " + colIx + ": " + from + "; " + to);
}
- LlapDataBuffer[][] fromColData = from.data[colIx];
+ LlapSerDeDataBuffer[][] fromColData = from.data[colIx];
if (fromColData != null) {
if (to.data[colIx] != null) {
// Note: we assume here that the data that was returned to the caller from cache will not
@@ -616,12 +623,22 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
}
}
- private void unlockBuffer(LlapDataBuffer buffer, boolean handleLastDecRef) {
+ @Override
+ public void decRefBuffer(MemoryBuffer buffer) {
+ unlockBuffer((LlapSerDeDataBuffer)buffer, true);
+ }
+
+ @Override
+ public void decRefBuffers(List<MemoryBuffer> cacheBuffers) {
+ for (MemoryBuffer b : cacheBuffers) {
+ unlockBuffer((LlapSerDeDataBuffer)b, true);
+ }
+ }
+
+ private void unlockBuffer(LlapSerDeDataBuffer buffer, boolean handleLastDecRef) {
boolean isLastDecref = (buffer.decRef() == 0);
if (handleLastDecRef && isLastDecref) {
- // This is kind of not pretty, but this is how we detect whether buffer was cached.
- // We would always set this for lookups at put time.
- if (buffer.declaredCachedLength != LlapDataBuffer.UNKNOWN_CACHED_LENGTH) {
+ if (buffer.isCached) {
cachePolicy.notifyUnlock(buffer);
} else {
if (LlapIoImpl.CACHE_LOGGER.isTraceEnabled()) {
@@ -633,13 +650,6 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
metrics.decrCacheNumLockedBuffers();
}
- private static final ByteBuffer fakeBuf = ByteBuffer.wrap(new byte[1]);
- public static LlapDataBuffer allocateFake() {
- LlapDataBuffer fake = new LlapDataBuffer();
- fake.initialize(-1, fakeBuf, 0, 1);
- return fake;
- }
-
public final void notifyEvicted(MemoryBuffer buffer) {
newEvictions.incrementAndGet();
}
@@ -664,14 +674,14 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
try {
for (StripeData sd : fd.stripes) {
for (int colIx = 0; colIx < sd.data.length; ++colIx) {
- LlapDataBuffer[][] colData = sd.data[colIx];
+ LlapSerDeDataBuffer[][] colData = sd.data[colIx];
if (colData == null) continue;
boolean hasAllData = true;
for (int j = 0; (j < colData.length) && hasAllData; ++j) {
- LlapDataBuffer[] streamData = colData[j];
+ LlapSerDeDataBuffer[] streamData = colData[j];
if (streamData == null) continue;
for (int k = 0; k < streamData.length; ++k) {
- LlapDataBuffer buf = streamData[k];
+ LlapSerDeDataBuffer buf = streamData[k];
hasAllData = hasAllData && lockBuffer(buf, false);
if (!hasAllData) break;
unlockBuffer(buf, true);
@@ -691,6 +701,18 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
}
@Override
+ public boolean incRefBuffer(MemoryBuffer buffer) {
+ // notifyReused implies that buffer is already locked; it's also called once for new
+ // buffers that are not cached yet. Don't notify cache policy.
+ return lockBuffer(((LlapSerDeDataBuffer)buffer), false);
+ }
+
+ @Override
+ public Allocator getAllocator() {
+ return allocator;
+ }
+
+ @Override
public String debugDumpForOom() {
StringBuilder sb = new StringBuilder("File cache state ");
for (Map.Entry<Object, FileCache<FileData>> e : cache.entrySet()) {
@@ -711,25 +733,29 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
@Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nSerDe cache state ");
- int allLocked = 0, allUnlocked = 0, allEvicted = 0;
+ int allLocked = 0, allUnlocked = 0, allEvicted = 0, allMoving = 0;
for (Map.Entry<Object, FileCache<FileData>> e : cache.entrySet()) {
if (!e.getValue().incRef()) continue;
try {
FileData fd = e.getValue().getCache();
- int fileLocked = 0, fileUnlocked = 0, fileEvicted = 0;
+ int fileLocked = 0, fileUnlocked = 0, fileEvicted = 0, fileMoving = 0;
sb.append(fd.colCount).append(" columns, ").append(fd.stripes.size()).append(" stripes; ");
for (StripeData stripe : fd.stripes) {
if (stripe.data == null) continue;
for (int i = 0; i < stripe.data.length; ++i) {
- LlapDataBuffer[][] colData = stripe.data[i];
+ LlapSerDeDataBuffer[][] colData = stripe.data[i];
if (colData == null) continue;
for (int j = 0; j < colData.length; ++j) {
- LlapDataBuffer[] streamData = colData[j];
+ LlapSerDeDataBuffer[] streamData = colData[j];
if (streamData == null) continue;
for (int k = 0; k < streamData.length; ++k) {
int newRc = streamData[k].incRef();
if (newRc < 0) {
- ++fileEvicted;
+ if (newRc == LlapAllocatorBuffer.INCREF_EVICTED) {
+ ++fileEvicted;
+ } else if (newRc == LlapAllocatorBuffer.INCREF_FAILED) {
+ ++fileMoving;
+ }
continue;
}
try {
@@ -748,13 +774,14 @@ public class SerDeLowLevelCacheImpl implements LlapOomDebugDump {
allLocked += fileLocked;
allUnlocked += fileUnlocked;
allEvicted += fileEvicted;
- sb.append("\n file " + e.getKey() + ": " + fileLocked + " locked, "
- + fileUnlocked + " unlocked, " + fileEvicted + " evicted");
+ allMoving += fileMoving;
+ sb.append("\n file " + e.getKey() + ": " + fileLocked + " locked, " + fileUnlocked
+ + " unlocked, " + fileEvicted + " evicted, " + fileMoving + " being moved");
} finally {
e.getValue().decRef();
}
}
- sb.append("\nSerDe cache summary: " + allLocked + " locked, "
- + allUnlocked + " unlocked, " + allEvicted + " evicted");
+ sb.append("\nSerDe cache summary: " + allLocked + " locked, " + allUnlocked + " unlocked, "
+ + allEvicted + " evicted, " + allMoving + " being moved");
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/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 51eb34e..f5a5f53 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
@@ -50,23 +50,32 @@ public final class SimpleAllocator implements Allocator, BuddyAllocatorMXBean {
}
}
+
@Override
+ @Deprecated
public void allocateMultiple(MemoryBuffer[] dest, int size) {
+ allocateMultiple(dest, size, null);
+ }
+
+ @Override
+ public void allocateMultiple(MemoryBuffer[] dest, int size, BufferObjectFactory factory) {
for (int i = 0; i < dest.length; ++i) {
- LlapDataBuffer buf = null;
+ LlapAllocatorBuffer buf = null;
if (dest[i] == null) {
- dest[i] = buf = createUnallocated();
+ // Note: this is backward compat only. Should be removed with createUnallocated.
+ dest[i] = buf = (factory != null)
+ ? (LlapAllocatorBuffer)factory.create() : createUnallocated();
} else {
- buf = (LlapDataBuffer)dest[i];
+ buf = (LlapAllocatorBuffer)dest[i];
}
ByteBuffer bb = isDirect ? ByteBuffer.allocateDirect(size) : ByteBuffer.allocate(size);
- buf.initialize(0, bb, 0, size);
+ buf.initialize(bb, 0, size);
}
}
@Override
public void deallocate(MemoryBuffer buffer) {
- LlapDataBuffer buf = (LlapDataBuffer)buffer;
+ LlapAllocatorBuffer buf = (LlapAllocatorBuffer)buffer;
ByteBuffer bb = buf.byteBuffer;
buf.byteBuffer = null;
if (!bb.isDirect()) return;
@@ -86,7 +95,8 @@ public final class SimpleAllocator implements Allocator, BuddyAllocatorMXBean {
}
@Override
- public LlapDataBuffer createUnallocated() {
+ @Deprecated
+ public LlapAllocatorBuffer createUnallocated() {
return new LlapDataBuffer();
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
index af7cf3d..076c80f 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
@@ -38,14 +38,14 @@ public class SimpleBufferManager implements BufferUsageManager, LowLevelCache {
this.metrics = metrics;
}
- private boolean lockBuffer(LlapDataBuffer buffer) {
+ private boolean lockBuffer(LlapAllocatorBuffer buffer) {
int rc = buffer.incRef();
if (rc <= 0) return false;
metrics.incrCacheNumLockedBuffers();
return true;
}
- private void unlockBuffer(LlapDataBuffer buffer) {
+ private void unlockBuffer(LlapAllocatorBuffer buffer) {
if (buffer.decRef() == 0) {
if (LlapIoImpl.CACHE_LOGGER.isTraceEnabled()) {
LlapIoImpl.CACHE_LOGGER.trace("Deallocating {} that was not cached", buffer);
@@ -57,19 +57,19 @@ public class SimpleBufferManager implements BufferUsageManager, LowLevelCache {
@Override
public void decRefBuffer(MemoryBuffer buffer) {
- unlockBuffer((LlapDataBuffer)buffer);
+ unlockBuffer((LlapAllocatorBuffer)buffer);
}
@Override
public void decRefBuffers(List<MemoryBuffer> cacheBuffers) {
for (MemoryBuffer b : cacheBuffers) {
- unlockBuffer((LlapDataBuffer)b);
+ unlockBuffer((LlapAllocatorBuffer)b);
}
}
@Override
public boolean incRefBuffer(MemoryBuffer buffer) {
- return lockBuffer((LlapDataBuffer)buffer);
+ return lockBuffer((LlapAllocatorBuffer)buffer);
}
@Override
@@ -88,7 +88,7 @@ public class SimpleBufferManager implements BufferUsageManager, LowLevelCache {
MemoryBuffer[] chunks, long baseOffset, Priority priority,
LowLevelCacheCounters qfCounters) {
for (int i = 0; i < chunks.length; ++i) {
- LlapDataBuffer buffer = (LlapDataBuffer)chunks[i];
+ LlapAllocatorBuffer buffer = (LlapAllocatorBuffer)chunks[i];
if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
LlapIoImpl.LOCKING_LOGGER.trace("Locking {} at put time (no cache)", buffer);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
index 253532a..d107e67 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
@@ -116,7 +116,7 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
OrcMetadataCache metadataCache = null;
LowLevelCache cache = null;
SerDeLowLevelCacheImpl serdeCache = null; // TODO: extract interface when needed
- BufferUsageManager bufferManager = null;
+ BufferUsageManager bufferManagerOrc = null, bufferManagerGeneric = null;
boolean isEncodeEnabled = HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_ENCODE_ENABLED);
if (useLowLevelCache) {
// Memory manager uses cache policy to trigger evictions, so create the policy first.
@@ -172,12 +172,13 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
cachePolicy.setParentDebugDumper(e);
cacheImpl.startThreads(); // Start the cache threads.
- bufferManager = cacheImpl; // Cache also serves as buffer manager.
+ bufferManagerOrc = cacheImpl; // Cache also serves as buffer manager.
+ bufferManagerGeneric = serdeCache;
} else {
this.allocator = new SimpleAllocator(conf);
memoryDumpRoot = null;
SimpleBufferManager sbm = new SimpleBufferManager(allocator, cacheMetrics);
- bufferManager = sbm;
+ bufferManagerOrc = bufferManagerGeneric = sbm;
cache = sbm;
}
// IO thread pool. Listening is used for unhandled errors for now (TODO: remove?)
@@ -189,9 +190,9 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
// TODO: this should depends on input format and be in a map, or something.
FixedSizedObjectPool<IoTrace> tracePool = IoTrace.createTracePool(conf);
this.orcCvp = new OrcColumnVectorProducer(
- metadataCache, cache, bufferManager, conf, cacheMetrics, ioMetrics, tracePool);
+ metadataCache, cache, bufferManagerOrc, conf, cacheMetrics, ioMetrics, tracePool);
this.genericCvp = isEncodeEnabled ? new GenericColumnVectorProducer(
- serdeCache, bufferManager, conf, cacheMetrics, ioMetrics, tracePool) : null;
+ serdeCache, bufferManagerGeneric, conf, cacheMetrics, ioMetrics, tracePool) : null;
LOG.info("LLAP IO initialized");
registerMXBeans();
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
index 655ce83..edd498e 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/OrcEncodedDataReader.java
@@ -40,6 +40,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.Pool;
import org.apache.hadoop.hive.common.Pool.PoolObjectHelper;
+import org.apache.hadoop.hive.common.io.Allocator.BufferObjectFactory;
import org.apache.hadoop.hive.common.io.DataCache;
import org.apache.hadoop.hive.common.io.Allocator;
import org.apache.hadoop.hive.common.io.encoded.EncodedColumnBatch.ColumnStreamData;
@@ -51,6 +52,7 @@ import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.llap.ConsumerFeedback;
import org.apache.hadoop.hive.llap.DebugUtils;
import org.apache.hadoop.hive.llap.cache.BufferUsageManager;
+import org.apache.hadoop.hive.llap.cache.LlapDataBuffer;
import org.apache.hadoop.hive.llap.cache.LowLevelCache;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters;
@@ -831,7 +833,7 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
stripeRgs = new boolean[stripeIxTo - stripeIxFrom][];
}
- private class DataWrapperForOrc implements DataReader, DataCache {
+ private class DataWrapperForOrc implements DataReader, DataCache, BufferObjectFactory {
private final DataReader orcDataReader;
private DataWrapperForOrc(DataWrapperForOrc other) {
@@ -951,6 +953,16 @@ public class OrcEncodedDataReader extends CallableWithNdc<Void>
public OrcProto.StripeFooter readStripeFooter(StripeInformation stripe) throws IOException {
return orcDataReader.readStripeFooter(stripe);
}
+
+ @Override
+ public BufferObjectFactory getDataBufferFactory() {
+ return this;
+ }
+
+ @Override
+ public MemoryBuffer create() {
+ return new LlapDataBuffer();
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
index 35d6178..187fa7d 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.Pool.PoolObjectHelper;
import org.apache.hadoop.hive.common.io.Allocator;
+import org.apache.hadoop.hive.common.io.Allocator.BufferObjectFactory;
import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
import org.apache.hadoop.hive.common.io.DiskRangeList;
import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
@@ -46,10 +47,12 @@ import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.llap.ConsumerFeedback;
import org.apache.hadoop.hive.llap.DebugUtils;
import org.apache.hadoop.hive.llap.cache.BufferUsageManager;
-import org.apache.hadoop.hive.llap.cache.LlapDataBuffer;
+import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
+import org.apache.hadoop.hive.llap.cache.LlapAllocatorBuffer;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl;
import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl.FileData;
+import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl.LlapSerDeDataBuffer;
import org.apache.hadoop.hive.llap.cache.SerDeLowLevelCacheImpl.StripeData;
import org.apache.hadoop.hive.llap.counters.LlapIOCounters;
import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters;
@@ -94,7 +97,6 @@ import org.apache.orc.OrcFile.Version;
import org.apache.orc.OrcProto;
import org.apache.orc.OrcProto.ColumnEncoding;
import org.apache.orc.TypeDescription;
-import org.apache.orc.impl.OutStream;
import org.apache.orc.PhysicalWriter;
import org.apache.orc.PhysicalWriter.OutputReceiver;
import org.apache.orc.impl.SchemaEvolution;
@@ -151,6 +153,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private final SerDeLowLevelCacheImpl cache;
private final BufferUsageManager bufferManager;
+ private final BufferObjectFactory bufferFactory;
private final Configuration daemonConf;
private final FileSplit split;
private List<Integer> columnIds;
@@ -191,6 +194,12 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
throws IOException {
this.cache = cache;
this.bufferManager = bufferManager;
+ this.bufferFactory = new BufferObjectFactory() {
+ @Override
+ public MemoryBuffer create() {
+ return new SerDeLowLevelCacheImpl.LlapSerDeDataBuffer();
+ }
+ };
this.parts = parts;
this.daemonConf = new Configuration(daemonConf);
// Disable dictionary encoding for the writer.
@@ -303,7 +312,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private static String toString(List<MemoryBuffer> data) {
String s = "";
for (MemoryBuffer buffer : data) {
- s += LlapDataBuffer.toDataString(buffer) + ", ";
+ s += buffer + ", ";
}
return s;
}
@@ -331,6 +340,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private CacheStripeData currentStripe;
private final List<CacheStripeData> stripes = new ArrayList<>();
private final BufferUsageManager bufferManager;
+ private final Allocator.BufferObjectFactory bufferFactory;
/**
* For !doesSourceHaveIncludes case, stores global column IDs to verify writer columns.
* For doesSourceHaveIncludes case, stores source column IDs used to map things.
@@ -344,13 +354,15 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private final AtomicBoolean isStopped;
public CacheWriter(BufferUsageManager bufferManager, List<Integer> columnIds,
- boolean[] writerIncludes, boolean doesSourceHaveIncludes, AtomicBoolean isStopped) {
+ boolean[] writerIncludes, boolean doesSourceHaveIncludes,
+ Allocator.BufferObjectFactory bufferFactory, AtomicBoolean isStopped) {
this.bufferManager = bufferManager;
assert writerIncludes != null; // Taken care of on higher level.
this.writerIncludes = writerIncludes;
this.doesSourceHaveIncludes = doesSourceHaveIncludes;
this.columnIds = columnIds;
this.isStopped = isStopped;
+ this.bufferFactory = bufferFactory;
startStripe();
}
@@ -438,7 +450,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
if (LlapIoImpl.LOG.isTraceEnabled()) {
LlapIoImpl.LOG.trace("Creating cache receiver for " + name);
}
- CacheOutputReceiver cor = new CacheOutputReceiver(bufferManager, name, isStopped);
+ CacheOutputReceiver cor = new CacheOutputReceiver(bufferManager, bufferFactory, name, isStopped);
or = cor;
List<CacheOutputReceiver> list = colStreams.get(name.getColumn());
if (list == null) {
@@ -568,6 +580,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private static final class CacheOutputReceiver implements CacheOutput, OutputReceiver {
private final BufferUsageManager bufferManager;
+ private final BufferObjectFactory bufferFactory;
private final StreamName name;
private List<MemoryBuffer> buffers = null;
private int lastBufferPos = -1;
@@ -576,8 +589,10 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private final StoppableAllocator allocator;
public CacheOutputReceiver(
- BufferUsageManager bufferManager, StreamName name, AtomicBoolean isStopped) {
+ BufferUsageManager bufferManager,
+ BufferObjectFactory bufferFactory, StreamName name, AtomicBoolean isStopped) {
this.bufferManager = bufferManager;
+ this.bufferFactory = bufferFactory;
Allocator alloc = bufferManager.getAllocator();
this.allocator = alloc instanceof StoppableAllocator ? (StoppableAllocator) alloc : null;
this.name = name;
@@ -598,13 +613,12 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private void allocateMultiple(MemoryBuffer[] dest, int size) {
if (allocator != null) {
- allocator.allocateMultiple(dest, size, isStopped);
+ allocator.allocateMultiple(dest, size, bufferFactory, isStopped);
} else {
- bufferManager.getAllocator().allocateMultiple(dest, size);
+ bufferManager.getAllocator().allocateMultiple(dest, size, bufferFactory);
}
}
-
@Override
public void output(ByteBuffer buffer) throws IOException {
// TODO: avoid put() by working directly in OutStream?
@@ -629,7 +643,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
if (isNewBuffer) {
MemoryBuffer[] dest = new MemoryBuffer[1];
allocateMultiple(dest, size);
- LlapDataBuffer newBuffer = (LlapDataBuffer)dest[0];
+ LlapSerDeDataBuffer newBuffer = (LlapSerDeDataBuffer)dest[0];
bb = newBuffer.getByteBufferRaw();
lastBufferPos = bb.position();
buffers.add(newBuffer);
@@ -720,10 +734,10 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private void unlockAllBuffers(StripeData si) {
for (int i = 0; i < si.getData().length; ++i) {
- LlapDataBuffer[][] colData = si.getData()[i];
+ LlapSerDeDataBuffer[][] colData = si.getData()[i];
if (colData == null) continue;
for (int j = 0; j < colData.length; ++j) {
- LlapDataBuffer[] streamData = colData[j];
+ LlapSerDeDataBuffer[] streamData = colData[j];
if (streamData == null) continue;
for (int k = 0; k < streamData.length; ++k) {
bufferManager.decRefBuffer(streamData[k]);
@@ -760,10 +774,10 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private void lockAllBuffers(StripeData sd) {
for (int i = 0; i < sd.getData().length; ++i) {
- LlapDataBuffer[][] colData = sd.getData()[i];
+ LlapSerDeDataBuffer[][] colData = sd.getData()[i];
if (colData == null) continue;
for (int j = 0; j < colData.length; ++j) {
- LlapDataBuffer[] streamData = colData[j];
+ LlapSerDeDataBuffer[] streamData = colData[j];
if (streamData == null) continue;
for (int k = 0; k < streamData.length; ++k) {
boolean canLock = bufferManager.incRefBuffer(streamData[k]);
@@ -937,7 +951,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
logProcessOneSlice(stripeIx, diskData, cacheData);
ColumnEncoding[] cacheEncodings = cacheData == null ? null : cacheData.getEncodings();
- LlapDataBuffer[][][] cacheBuffers = cacheData == null ? null : cacheData.getData();
+ LlapSerDeDataBuffer[][][] cacheBuffers = cacheData == null ? null : cacheData.getData();
long cacheRowCount = cacheData == null ? -1L : cacheData.getRowCount();
SerDeStripeMetadata metadata = new SerDeStripeMetadata(stripeIx);
StripeData sliceToCache = null;
@@ -963,7 +977,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
if (!hasAllData && splitIncludes[colIx]) {
// The column has been read from disk.
List<CacheWriter.CacheStreamData> streams = diskData.colStreams.get(colIx);
- LlapDataBuffer[][] newCacheDataForCol = createArrayToCache(sliceToCache, colIx, streams);
+ LlapSerDeDataBuffer[][] newCacheDataForCol = createArrayToCache(sliceToCache, colIx, streams);
if (streams == null) continue; // Struct column, such as root?
Iterator<CacheWriter.CacheStreamData> iter = streams.iterator();
while (iter.hasNext()) {
@@ -1026,7 +1040,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
return true; // Nothing to process.
}
ColumnEncoding[] cacheEncodings = cacheData == null ? null : cacheData.getEncodings();
- LlapDataBuffer[][][] cacheBuffers = cacheData == null ? null : cacheData.getData();
+ LlapSerDeDataBuffer[][][] cacheBuffers = cacheData == null ? null : cacheData.getData();
if (cacheData != null) {
// Don't validate column count - no encodings for vectors.
validateCacheAndDisk(cacheData, diskData.getRowCount(), -1, diskData);
@@ -1078,7 +1092,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
if (!splitIncludes[colIx]) continue;
// The column has been read from disk.
List<CacheWriter.CacheStreamData> streams = diskData.colStreams.get(colIx);
- LlapDataBuffer[][] newCacheDataForCol = createArrayToCache(sliceToCache, colIx, streams);
+ LlapSerDeDataBuffer[][] newCacheDataForCol = createArrayToCache(sliceToCache, colIx, streams);
if (streams == null) continue; // Struct column, such as root?
Iterator<CacheWriter.CacheStreamData> iter = streams.iterator();
while (iter.hasNext()) {
@@ -1132,28 +1146,28 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
}
- private static LlapDataBuffer[][] createArrayToCache(
+ private static LlapSerDeDataBuffer[][] createArrayToCache(
StripeData sliceToCache, int colIx, List<CacheWriter.CacheStreamData> streams) {
if (LlapIoImpl.LOG.isTraceEnabled()) {
LlapIoImpl.LOG.trace("Processing streams for column " + colIx + ": " + streams);
}
- LlapDataBuffer[][] newCacheDataForCol = sliceToCache.getData()[colIx]
- = new LlapDataBuffer[OrcEncodedColumnBatch.MAX_DATA_STREAMS][];
+ LlapSerDeDataBuffer[][] newCacheDataForCol = sliceToCache.getData()[colIx]
+ = new LlapSerDeDataBuffer[OrcEncodedColumnBatch.MAX_DATA_STREAMS][];
return newCacheDataForCol;
}
private static int setStreamDataToCache(
- LlapDataBuffer[][] newCacheDataForCol, CacheWriter.CacheStreamData stream) {
+ LlapSerDeDataBuffer[][] newCacheDataForCol, CacheWriter.CacheStreamData stream) {
int streamIx = stream.name.getKind().getNumber();
- // This is kinda hacky - we "know" these are LlapDataBuffer-s.
- newCacheDataForCol[streamIx] = stream.data.toArray(new LlapDataBuffer[stream.data.size()]);
+ // This is kinda hacky - we "know" these are LlaSerDeDataBuffer-s.
+ newCacheDataForCol[streamIx] = stream.data.toArray(new LlapSerDeDataBuffer[stream.data.size()]);
return streamIx;
}
- private void processColumnCacheData(LlapDataBuffer[][][] cacheBuffers,
+ private void processColumnCacheData(LlapSerDeDataBuffer[][][] cacheBuffers,
OrcEncodedColumnBatch ecb, int colIx) {
// The column has been obtained from cache.
- LlapDataBuffer[][] colData = cacheBuffers[colIx];
+ LlapSerDeDataBuffer[][] colData = cacheBuffers[colIx];
if (LlapIoImpl.CACHE_LOGGER.isTraceEnabled()) {
LlapIoImpl.CACHE_LOGGER.trace("Processing cache data for column " + colIx + ": "
+ SerDeLowLevelCacheImpl.toString(colData));
@@ -1178,8 +1192,6 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
private void discardUncachedBuffers(List<MemoryBuffer> list) {
for (MemoryBuffer buffer : list) {
- boolean isInvalidated = ((LlapDataBuffer)buffer).invalidate();
- assert isInvalidated;
bufferManager.getAllocator().deallocate(buffer);
}
}
@@ -1404,7 +1416,7 @@ public class SerDeEncodedDataReader extends CallableWithNdc<Void>
// TODO: move this into ctor? EW would need to create CacheWriter then
List<Integer> cwColIds = writer.isOnlyWritingIncludedColumns() ? splitColumnIds : columnIds;
writer.init(new CacheWriter(bufferManager, cwColIds, splitIncludes,
- writer.isOnlyWritingIncludedColumns(), isStopped), daemonConf, split.getPath());
+ writer.isOnlyWritingIncludedColumns(), bufferFactory, isStopped), daemonConf, split.getPath());
if (writer instanceof VectorDeserializeOrcWriter) {
VectorDeserializeOrcWriter asyncWriter = (VectorDeserializeOrcWriter)writer;
asyncWriter.startAsync(new AsyncCacheDataCallback());
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VectorDeserializeOrcWriter.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VectorDeserializeOrcWriter.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VectorDeserializeOrcWriter.java
index c9df7d9..981e52f 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VectorDeserializeOrcWriter.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VectorDeserializeOrcWriter.java
@@ -228,6 +228,9 @@ class VectorDeserializeOrcWriter extends EncodingWriter implements Runnable {
WriteOperation op = null;
int fallbackMs = 8;
while (true) {
+ // The reason we poll here is that a blocking queue causes the query thread to spend
+ // non-trivial amount of time signaling when an element is added; we'd rather that the
+ // time was wasted on this background thread.
op = queue.poll();
if (op != null) break;
if (fallbackMs > 262144) { // Arbitrary... we don't expect caller to hang out for 7+ mins.
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
index 20af0d0..dc053ee 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileEstimateErrors.java
@@ -104,8 +104,8 @@ public class OrcFileEstimateErrors extends LlapCacheableBuffer {
}
@Override
- protected boolean invalidate() {
- return true;
+ protected int invalidate() {
+ return INVALIDATE_OK;
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
index 2c7a234..b9d7a77 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcFileMetadata.java
@@ -131,8 +131,8 @@ public final class OrcFileMetadata extends LlapCacheableBuffer
}
@Override
- protected boolean invalidate() {
- return true; // relies on GC, so it can always be evicted now.
+ protected int invalidate() {
+ return INVALIDATE_OK; // relies on GC, so it can always be evicted now.
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/bd32deb4/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
index 1f3f7ea..17266bc 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/OrcStripeMetadata.java
@@ -168,8 +168,8 @@ public class OrcStripeMetadata extends LlapCacheableBuffer implements ConsumerSt
}
@Override
- protected boolean invalidate() {
- return true;
+ protected int invalidate() {
+ return INVALIDATE_OK;
}
@Override