You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by sz...@apache.org on 2020/11/11 13:36:39 UTC
[hive] branch master updated: HIVE-24340: Mark phase for proactive
cache eviction (Adam Szita, reviewed by Peter Vary and Antal Sinkovits)
This is an automated email from the ASF dual-hosted git repository.
szita pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 0cd38cb HIVE-24340: Mark phase for proactive cache eviction (Adam Szita, reviewed by Peter Vary and Antal Sinkovits)
0cd38cb is described below
commit 0cd38cbda57a06d207c70827936aa319c67d9ef2
Author: Adam Szita <40...@users.noreply.github.com>
AuthorDate: Wed Nov 11 14:36:19 2020 +0100
HIVE-24340: Mark phase for proactive cache eviction (Adam Szita, reviewed by Peter Vary and Antal Sinkovits)
---
.../hive/llap/cache/LlapAllocatorBuffer.java | 52 +++++++++++++++++---
.../hive/llap/cache/LlapCacheableBuffer.java | 17 +++++++
.../hadoop/hive/llap/cache/LowLevelCache.java | 10 ++++
.../hadoop/hive/llap/cache/LowLevelCacheImpl.java | 42 ++++++++++++++--
.../llap/cache/LowLevelCacheMemoryManager.java | 16 +-----
.../hive/llap/cache/SerDeLowLevelCacheImpl.java | 46 +++++++++++++++--
.../hive/llap/cache/SimpleBufferManager.java | 6 +++
.../hadoop/hive/llap/io/api/impl/LlapIoImpl.java | 25 ++++++++--
.../hive/llap/io/metadata/MetadataCache.java | 25 ++++++++++
.../llap/io/metadata/OrcFileEstimateErrors.java | 15 ++++++
.../hive/llap/cache/TestLowLevelCacheImpl.java | 54 +++++++++++++++++++-
.../hive/llap/cache/TestOrcMetadataCache.java | 57 +++++++++++++++++++++-
.../hadoop/hive/common/io/FileMetadataCache.java | 9 ++++
13 files changed, 337 insertions(+), 37 deletions(-)
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
index 339335f..dafb6cf 100644
--- 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
@@ -233,6 +233,43 @@ public abstract class LlapAllocatorBuffer extends LlapCacheableBuffer implements
}
}
+ /**
+ * Marks this buffer as eligible for proactive eviction. No need to check refCount here.
+ * @return buffer size
+ */
+ @Override
+ public long markForEviction() {
+ long oldValue, newValue;
+ do {
+ oldValue = state.get();
+ if (State.hasFlags(oldValue, State.FLAG_EVICTED | State.FLAG_MARKED)) {
+ // Either got evicted concurrently, or was marked by another event already (should be very rare),
+ // nothing to do now.
+ return 0;
+ }
+ newValue = State.setFlag(oldValue, State.FLAG_MARKED);
+ } while (!state.compareAndSet(oldValue, newValue));
+ return getMemoryUsage();
+ }
+
+ @Override
+ public void removeProactiveEvictionMark() {
+ long oldValue, newValue;
+ do {
+ oldValue = state.get();
+ if (!State.hasFlags(oldValue, State.FLAG_MARKED)) {
+ return;
+ }
+ newValue = State.switchFlag(oldValue, State.FLAG_MARKED);
+ } while (!state.compareAndSet(oldValue, newValue));
+ }
+
+ @VisibleForTesting
+ @Override
+ public boolean isMarkedForEviction() {
+ return State.hasFlags(state.get(), State.FLAG_MARKED);
+ }
+
@Override
public String toString() {
long state = this.state.get();
@@ -339,13 +376,14 @@ public abstract class LlapAllocatorBuffer extends LlapCacheableBuffer implements
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;
+ FLAG_MOVING = 0b000001, // Locked by someone to move or force-evict.
+ FLAG_EVICTED = 0b000010, // Evicted. This is cache-specific.
+ FLAG_REMOVED = 0b000100, // Removed from allocator structures. The final state.
+ FLAG_MEM_RELEASED = 0b001000, // The memory was released to memory manager.
+ FLAG_NEW_ALLOC = 0b010000, // New allocation before the first use; cannot force-evict.
+ FLAG_MARKED = 0b100000; // Marked for eviction, but not evicted yet. Used for proactive eviction.
+ private static final int FLAGS_WIDTH = 6,
+ REFCOUNT_WIDTH = 18, ARENA_WIDTH = 16, HEADER_WIDTH = 24;
public static final long MAX_REFCOUNT = (1 << REFCOUNT_WIDTH) - 1;
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 0526033..564a934 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
@@ -61,4 +61,21 @@ public abstract class LlapCacheableBuffer {
public abstract CacheTag getTag();
protected abstract boolean isLocked();
+
+ /**
+ * Marks this buffer as eligible for proactive eviction.
+ * @return buffer size
+ */
+ public abstract long markForEviction();
+
+ /**
+ * Un-marks proactive eviction flag from this buffer.
+ */
+ public abstract void removeProactiveEvictionMark();
+
+ /**
+ * Checks if this buffer is marked for proactive eviction
+ * @return true if marked
+ */
+ public abstract boolean isMarkedForEviction();
}
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java
index f780b29..0c83238 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hive.llap.cache;
+import java.util.function.Predicate;
+
import org.apache.hadoop.hive.common.io.CacheTag;
import org.apache.hadoop.hive.common.io.DiskRange;
import org.apache.hadoop.hive.common.io.DiskRangeList;
@@ -63,4 +65,12 @@ public interface LowLevelCache {
/** Notifies the cache that a particular buffer should be removed due to eviction. */
void notifyEvicted(MemoryBuffer buffer);
+
+ /**
+ * Iterates through the file entries of this cache and for those that match the given predicate (aka have a matching
+ * CacheTag) will have their buffers marked for (a later) proactive eviction.
+ * @param predicate - matching the predicate indicates eligibility for proactive eviction
+ * @return number of bytes marked in the buffers eligible for eviction
+ */
+ long markBuffersForProactiveEviction(Predicate<CacheTag> predicate);
}
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 59fedbd..89602a6 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
@@ -19,12 +19,14 @@ package org.apache.hadoop.hive.llap.cache;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
import org.apache.hadoop.hive.common.io.Allocator;
import org.apache.hadoop.hive.common.io.CacheTag;
@@ -176,6 +178,28 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
return prev.next;
}
+ @Override
+ public long markBuffersForProactiveEviction(Predicate<CacheTag> predicate) {
+ long markedBytes = 0;
+ // Proactive eviction does not need to be perfectly accurate - the iterator returned here might be missing some
+ // concurrent inserts / removals but it's fine for us here.
+ Collection<FileCache<ConcurrentSkipListMap<Long, LlapDataBuffer>>> fileCaches = cache.values();
+ for (FileCache<ConcurrentSkipListMap<Long, LlapDataBuffer>> fileCache : fileCaches) {
+ if (predicate.test(fileCache.getTag()) && fileCache.incRef()) {
+ // Locked on subcache so that the async cleaner won't pull out buffers in an unlikely scenario.
+ try {
+ Collection<LlapDataBuffer> buffers = fileCache.getCache().values();
+ for (LlapDataBuffer buffer : buffers) {
+ markedBytes += buffer.markForEviction();
+ }
+ } finally {
+ fileCache.decRef();
+ }
+ }
+ }
+ return markedBytes;
+ }
+
private void getOverlappingRanges(long baseOffset, DiskRangeList currentNotCached,
ConcurrentSkipListMap<Long, LlapDataBuffer> cache, DiskRangeListFactory factory,
BooleanRef gotAllData) {
@@ -455,13 +479,13 @@ 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, allMoving = 0;
+ int allLocked = 0, allUnlocked = 0, allEvicted = 0, allMoving = 0, allMarked = 0;
long totalUsedSpace = 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, fileMoving = 0;
+ int fileLocked = 0, fileUnlocked = 0, fileEvicted = 0, fileMoving = 0, fileMarked = 0;
long fileMemoryUsage = 0;
if (e.getValue().getCache().isEmpty()) continue;
List<LlapDataBuffer> lockedBufs = null;
@@ -478,6 +502,9 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
}
continue;
}
+ if (e2.getValue().isMarkedForEviction()) {
+ ++fileMarked;
+ }
try {
if (newRc > 1) { // We hold one refcount.
++fileLocked;
@@ -496,6 +523,7 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
allUnlocked += fileUnlocked;
allEvicted += fileEvicted;
allMoving += fileMoving;
+ allMarked += fileMarked;
totalUsedSpace += fileMemoryUsage;
sb.append("\n file "
@@ -508,7 +536,9 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
+ fileEvicted
+ " evicted, "
+ fileMoving
- + " being moved,"
+ + " being moved, "
+ + fileMarked
+ + " marked for eviction, "
+ fileMemoryUsage
+ " total used byte");
if (fileLocked > 0 && LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
@@ -526,8 +556,10 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
+ allEvicted
+ " evicted, "
+ allMoving
- + " being moved,"
+ + " being moved, "
+ + allMarked
+ + " marked for eviction, "
+ totalUsedSpace
- + "total used space");
+ + " total used space");
}
}
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 afe1082..87fca68 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
@@ -166,20 +166,8 @@ public class LowLevelCacheMemoryManager implements MemoryManager {
return evicted;
}
- public long evictEntity(Predicate<LlapCacheableBuffer> predicate) {
- if (evictor == null) {
- return 0;
- }
- long evicted = evictor.evictEntity(predicate);
- if (evicted == 0) {
- return 0;
- }
- long usedMem = -1;
- do {
- usedMem = usedMemory.get();
- } while (!usedMemory.compareAndSet(usedMem, usedMem - evicted));
- metrics.incrCacheCapacityUsed(-evicted);
- return evicted;
+ public void notifyProactiveEvictionMark() {
+ // TODO: forward this signal to the cache policy if it supports proactive eviction
}
@VisibleForTesting
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 371eeb3..f890d12 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
@@ -23,6 +23,7 @@ import com.google.common.base.Function;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
@@ -31,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Predicate;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
@@ -516,6 +518,37 @@ public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapIoDebugDu
return rc > 0;
}
+ public long markBuffersForProactiveEviction(Predicate<CacheTag> predicate) {
+ long markedBytes = 0;
+ // Proactive eviction does not need to be perfectly accurate - the iterator returned here might be missing some
+ // concurrent inserts / removals but it's fine for us here.
+ Collection<FileCache<FileData>> fileCaches = cache.values();
+ for (FileCache<FileData> fileCache : fileCaches) {
+ if (predicate.test(fileCache.getTag()) && fileCache.incRef()) {
+ // Locked on subcache so that the async cleaner won't pull out buffers in an unlikely scenario.
+ try {
+ ArrayList<StripeData> stripeDataList = fileCache.getCache().getData();
+ for (StripeData stripeData : stripeDataList) {
+ for (int i = 0; i < stripeData.getData().length; ++i) {
+ LlapSerDeDataBuffer[][] colData = stripeData.getData()[i];
+ if (colData == null) continue;
+ for (int j = 0; j < colData.length; ++j) {
+ LlapSerDeDataBuffer[] streamData = colData[j];
+ if (streamData == null) continue;
+ for (int k = 0; k < streamData.length; ++k) {
+ markedBytes += streamData[k].markForEviction();
+ }
+ }
+ }
+ }
+ } finally {
+ fileCache.decRef();
+ }
+ }
+ }
+ return markedBytes;
+ }
+
public void putFileData(final FileData data, Priority priority,
LowLevelCacheCounters qfCounters, CacheTag tag) {
// TODO: buffers are accounted for at allocation time, but ideally we should report the memory
@@ -757,12 +790,12 @@ public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapIoDebugDu
@Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nSerDe cache state ");
- int allLocked = 0, allUnlocked = 0, allEvicted = 0, allMoving = 0;
+ int allLocked = 0, allUnlocked = 0, allEvicted = 0, allMoving = 0, allMarked = 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, fileMoving = 0;
+ int fileLocked = 0, fileUnlocked = 0, fileEvicted = 0, fileMoving = 0, fileMarked = 0;
for (StripeData stripe : fd.stripes) {
if (stripe.data == null) continue;
for (int i = 0; i < stripe.data.length; ++i) {
@@ -781,6 +814,9 @@ public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapIoDebugDu
}
continue;
}
+ if (streamData[k].isMarkedForEviction()) {
+ ++fileMarked;
+ }
try {
if (newRc > 1) { // We hold one refcount.
++fileLocked;
@@ -798,14 +834,16 @@ public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapIoDebugDu
allUnlocked += fileUnlocked;
allEvicted += fileEvicted;
allMoving += fileMoving;
+ allMarked += fileMarked;
sb.append("\n file " + e.getKey() + ": " + fileLocked + " locked, " + fileUnlocked
- + " unlocked, " + fileEvicted + " evicted, " + fileMoving + " being moved; ");
+ + " unlocked, " + fileEvicted + " evicted, " + fileMoving + " being moved, " + fileMarked
+ + " marked for eviction; ");
sb.append(fd.colCount).append(" columns, ").append(fd.stripes.size()).append(" stripes");
} finally {
e.getValue().decRef();
}
}
sb.append("\nSerDe cache summary: " + allLocked + " locked, " + allUnlocked + " unlocked, "
- + allEvicted + " evicted, " + allMoving + " being moved");
+ + allEvicted + " evicted, " + allMoving + " being moved, " + allMarked + " marked for eviction");
}
}
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 b2f606a..85772fa 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
@@ -18,6 +18,7 @@
package org.apache.hadoop.hive.llap.cache;
import java.util.List;
+import java.util.function.Predicate;
import org.apache.hadoop.hive.common.io.Allocator;
import org.apache.hadoop.hive.common.io.CacheTag;
@@ -102,4 +103,9 @@ public class SimpleBufferManager implements BufferUsageManager, LowLevelCache {
public void notifyEvicted(MemoryBuffer buffer) {
throw new UnsupportedOperationException("Buffer manager doesn't have cache");
}
+
+ @Override
+ public long markBuffersForProactiveEviction(Predicate<CacheTag> predicate) {
+ throw new UnsupportedOperationException("Buffer manager doesn't have cache");
+ }
}
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 8f27ca1..cbc87c8 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
@@ -103,6 +103,7 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch>, LlapIoDebugDump {
private final Allocator allocator;
private final FileMetadataCache fileMetadataCache;
private final LowLevelCache dataCache;
+ private final SerDeLowLevelCacheImpl serdeCache;
private final BufferUsageManager bufferManager;
private final Configuration daemonConf;
private final LowLevelCacheMemoryManager memoryManager;
@@ -150,6 +151,8 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch>, LlapIoDebugDump {
LowLevelCachePolicy
realCachePolicy =
useLrfu ? new LowLevelLrfuCachePolicy(minAllocSize, totalMemorySize, conf) : new LowLevelFifoCachePolicy();
+ // TODO: if realCachePolicy is not something that supports proactive caching
+ // turn the feature off (LLAP_IO_PROACTIVE_EVICTION_ENABLED) and log it
boolean trackUsage = HiveConf.getBoolVar(conf, HiveConf.ConfVars.LLAP_TRACK_CACHE_USAGE);
LowLevelCachePolicy cachePolicyWrapper;
if (trackUsage) {
@@ -212,6 +215,7 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch>, LlapIoDebugDump {
}
});
}
+ this.serdeCache = serdeCache;
// IO thread pool. Listening is used for unhandled errors for now (TODO: remove?)
int numThreads = HiveConf.getIntVar(conf, HiveConf.ConfVars.LLAP_IO_THREADPOOL_SIZE);
executor = new StatsRecordingThreadPool(numThreads, numThreads, 0L, TimeUnit.MILLISECONDS,
@@ -253,18 +257,31 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch>, LlapIoDebugDump {
}
final ProactiveEviction.Request request = ProactiveEviction.Request.Builder.create()
.fromProtoRequest(protoRequest).build();
- Predicate<LlapCacheableBuffer> predicate = buffer -> request.isTagMatch(buffer.getTag());
+ Predicate<CacheTag> predicate = tag -> request.isTagMatch(tag);
LOG.debug("Starting proactive eviction.");
- long evictedBytes = memoryManager.evictEntity(predicate);
+ long time = System.currentTimeMillis();
+
+ long markedBytes = dataCache.markBuffersForProactiveEviction(predicate);
+ markedBytes += fileMetadataCache.markBuffersForProactiveEviction(predicate);
+ markedBytes += serdeCache.markBuffersForProactiveEviction(predicate);
+
+ // Signal mark phase of proactive eviction was done
+ if (markedBytes > 0) {
+ memoryManager.notifyProactiveEvictionMark();
+ }
+
+ time = System.currentTimeMillis() - time;
+
if (LOG.isDebugEnabled()) {
StringBuilder sb = new StringBuilder();
- sb.append("Evicted ").append(evictedBytes).append(" bytes from LLAP cache buffers that belong to table(s): ");
+ sb.append(markedBytes).append(" bytes marked for eviction from LLAP cache buffers that belong to table(s): ");
for (String table : request.getEntities().get(request.getSingleDbName()).keySet()) {
sb.append(table).append(" ");
}
+ sb.append(" Duration: ").append(time).append(" ms");
LOG.debug(sb.toString());
}
- return evictedBytes;
+ return markedBytes;
}
@Override
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
index ce1e236..978cfc3 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
@@ -27,8 +27,10 @@ import java.io.InputStream;
import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers;
import java.nio.ByteBuffer;
+import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
import org.apache.hadoop.hive.common.io.DiskRange;
import org.apache.hadoop.hive.common.io.DiskRangeList;
@@ -123,6 +125,29 @@ public class MetadataCache implements LlapIoDebugDump, FileMetadataCache {
estimateErrors.remove(buffer.getFileKey());
}
+ @Override
+ public long markBuffersForProactiveEviction(Predicate<CacheTag> predicate) {
+ long markedBytes = 0;
+
+ Collection<LlapBufferOrBuffers> metadataBufferGroups = metadata.values();
+ for (LlapBufferOrBuffers bufferOrBuffers : metadataBufferGroups) {
+ LlapAllocatorBuffer singleBuffer = bufferOrBuffers.getSingleLlapBuffer();
+ if (singleBuffer != null) {
+ if (predicate.test(singleBuffer.getTag())) {
+ markedBytes += singleBuffer.markForEviction();
+ }
+ continue;
+ }
+ LlapAllocatorBuffer[] buffers = bufferOrBuffers.getMultipleLlapBuffers();
+ assert buffers != null && buffers.length > 1 : "Should have multiple buffers if NULL is set as single buffer";
+ if (predicate.test(buffers[0].getTag())) {
+ for (LlapAllocatorBuffer buffer : buffers) {
+ markedBytes += buffer.markForEviction();
+ }
+ }
+ }
+ return markedBytes;
+ }
@Override
public void debugDumpShort(StringBuilder sb) {
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 33e1680..e3cc386 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
@@ -124,6 +124,21 @@ public class OrcFileEstimateErrors extends LlapCacheableBuffer {
}
@Override
+ public long markForEviction() {
+ return 0;
+ }
+
+ @Override
+ public void removeProactiveEvictionMark() {
+ //No-op
+ }
+
+ @Override
+ public boolean isMarkedForEviction() {
+ return false;
+ }
+
+ @Override
public CacheTag getTag() {
// We don't care about these.
return CacheTag.build("OrcEstimates");
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
index 11a0128..de3e36c 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
@@ -17,7 +17,9 @@
*/
package org.apache.hadoop.hive.llap.cache;
+import static org.apache.hadoop.hive.llap.cache.LlapCacheableBuffer.INVALIDATE_OK;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
@@ -35,6 +37,8 @@ import java.util.stream.IntStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.hive.common.io.CacheTag;
import org.apache.hadoop.hive.common.io.DiskRange;
import org.apache.hadoop.hive.common.io.DiskRangeList;
import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
@@ -43,6 +47,7 @@ import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
import org.apache.hadoop.hive.ql.io.orc.encoded.CacheChunk;
+
import org.junit.Test;
public class TestLowLevelCacheImpl {
@@ -268,6 +273,51 @@ Example code to test specific scenarios:
}
@Test
+ public void testProactiveEvictionMark() {
+ LowLevelCacheImpl cache = new LowLevelCacheImpl(
+ LlapDaemonCacheMetrics.create("test", "1"), new DummyCachePolicy(),
+ new DummyAllocator(), true, -1); // no cleanup thread
+ long fn1 = 1;
+ long fn2 = 2;
+
+ LlapDataBuffer[] buffs1 = IntStream.range(0, 4).mapToObj(i -> fb()).toArray(LlapDataBuffer[]::new);
+ DiskRange[] drs1 = drs(IntStream.range(1, 5).toArray());
+ CacheTag tag1 = CacheTag.build("default.table1");
+
+ LlapDataBuffer[] buffs2 = IntStream.range(0, 41).mapToObj(i -> fb()).toArray(LlapDataBuffer[]::new);
+ DiskRange[] drs2 = drs(IntStream.range(1, 42).toArray());
+ CacheTag tag2 = CacheTag.build("default.table2");
+
+ Predicate<CacheTag> predicate = tag -> "default.table1".equals(tag.getTableName());
+
+ cache.putFileData(fn1, drs1, buffs1, 0, Priority.NORMAL, null, tag1);
+ cache.putFileData(fn2, drs2, buffs2, 0, Priority.NORMAL, null, tag2);
+
+ // Simulating eviction on some buffers
+ buffs1[2].decRef();
+ buffs1[2].decRef();
+ assertEquals(INVALIDATE_OK, buffs1[2].invalidate());
+
+ //buffs1[0,1,3] should be marked, as 2 is already invalidated
+ assertEquals(3, cache.markBuffersForProactiveEviction(predicate));
+
+ for (int i = 0; i < buffs1.length; ++i) {
+ LlapDataBuffer buffer = buffs1[i];
+ if (i == 2) {
+ assertFalse(buffer.isMarkedForEviction());
+ } else {
+ assertTrue(buffer.isMarkedForEviction());
+ }
+ }
+
+ // All buffers for file2 should not be marked as per predicate
+ for (LlapDataBuffer buffer : buffs2) {
+ assertFalse(buffer.isMarkedForEviction());
+ }
+
+ }
+
+ @Test
public void testMultiMatch() {
LowLevelCacheImpl cache = new LowLevelCacheImpl(
LlapDaemonCacheMetrics.create("test", "1"), new DummyCachePolicy(),
@@ -505,7 +555,7 @@ Example code to test specific scenarios:
if (r instanceof CacheChunk) {
LlapDataBuffer result = (LlapDataBuffer)((CacheChunk)r).getBuffer();
cache.decRefBuffer(result);
- if (victim == null && result.invalidate() == LlapCacheableBuffer.INVALIDATE_OK) {
+ if (victim == null && result.invalidate() == INVALIDATE_OK) {
++evictions;
victim = result;
}
@@ -558,7 +608,7 @@ Example code to test specific scenarios:
for (int i = 0; i < refCount; ++i) {
victimBuffer.decRef();
}
- assertTrue(LlapCacheableBuffer.INVALIDATE_OK == victimBuffer.invalidate());
+ assertTrue(INVALIDATE_OK == victimBuffer.invalidate());
cache.notifyEvicted(victimBuffer);
}
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
index 29aab30..848286f 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
@@ -17,11 +17,11 @@
*/
package org.apache.hadoop.hive.llap.cache;
+import static org.apache.hadoop.hive.llap.cache.LlapCacheableBuffer.INVALIDATE_OK;
import static org.junit.Assert.*;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
@@ -320,6 +320,61 @@ public class TestOrcMetadataCache {
OrcEncodedDataReader.getOrcTailForPath(path, conf, null, conf, null, null);
}
+ @Test
+ public void testProactiveEvictionMark() throws Exception {
+ DummyMemoryManager mm = new DummyMemoryManager();
+ DummyCachePolicy cp = new DummyCachePolicy();
+ final int MAX_ALLOC = 64;
+ LlapDaemonCacheMetrics metrics = LlapDaemonCacheMetrics.create("", "");
+ BuddyAllocator alloc = new BuddyAllocator(
+ false, false, 8, MAX_ALLOC, 1, 4096, 0, null, mm, metrics, null, true);
+ MetadataCache cache = new MetadataCache(alloc, mm, cp, true, metrics);
+
+ long fn1 = 1;
+ long fn2 = 2;
+ long fn3 = 3;
+
+ AtomicBoolean isStopped = new AtomicBoolean(false);
+
+ // Case for when metadata consists of just 1 buffer (most of the realworld cases)
+ ByteBuffer bb = ByteBuffer.wrap("small-meta-data-content".getBytes());
+ // Case for when metadata consists of multiple buffers (rare case), (max allocation is 64 hence the test data
+ // below is of length 65
+ ByteBuffer bb2 = ByteBuffer.wrap("-large-meta-data-content-large-meta-data-content-large-meta-data-".getBytes());
+
+ LlapBufferOrBuffers table1Buffers1 = cache.putFileMetadata(fn1, bb, CacheTag.build("default.table1"), isStopped);
+ assertNotNull(table1Buffers1.getSingleLlapBuffer());
+
+ LlapBufferOrBuffers table1Buffers2 = cache.putFileMetadata(fn2, bb2, CacheTag.build("default.table1"), isStopped);
+ assertNotNull(table1Buffers2.getMultipleLlapBuffers());
+ assertEquals(2, table1Buffers2.getMultipleLlapBuffers().length);
+
+ // Case for when metadata consists of just 1 buffer (most of the realworld cases)
+ ByteBuffer bb3 = ByteBuffer.wrap("small-meta-data-content-for-otherFile".getBytes());
+ LlapBufferOrBuffers table2Buffers1 = cache.putFileMetadata(fn3, bb3, CacheTag.build("default.table2"), isStopped);
+ assertNotNull(table2Buffers1.getSingleLlapBuffer());
+
+ Predicate<CacheTag> predicate = tag -> "default.table1".equals(tag.getTableName());
+
+ // Simulating eviction on some buffers
+ table1Buffers2.getMultipleLlapBuffers()[1].decRef();
+ assertEquals(INVALIDATE_OK, table1Buffers2.getMultipleLlapBuffers()[1].invalidate());
+
+ // table1Buffers1:27 (allocated as 32) + table1Buffers2[0]:64 (also allocated as 64)
+ assertEquals(96, cache.markBuffersForProactiveEviction(predicate));
+
+ // Single buffer for file1 should be marked as per predicate
+ assertTrue(table1Buffers1.getSingleLlapBuffer().isMarkedForEviction());
+
+ // Multi buffer for file2 should be partially marked as per predicate and prior eviction
+ assertTrue(table1Buffers2.getMultipleLlapBuffers()[0].isMarkedForEviction());
+ assertFalse(table1Buffers2.getMultipleLlapBuffers()[1].isMarkedForEviction());
+
+ // Single buffer for file3 should not be marked as per predicate
+ assertFalse(table2Buffers1.getSingleLlapBuffer().isMarkedForEviction());
+
+ }
+
private static final int INCOMPLETE = 0, DRL = 1;
public void verifyResult(DiskRangeList result, long... vals) {
for (int i = 0; i < vals.length; i += 3) {
diff --git a/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java b/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java
index 4691722..fe72fc5 100644
--- a/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java
+++ b/storage-api/src/java/org/apache/hadoop/hive/common/io/FileMetadataCache.java
@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
import java.io.IOException;
import java.io.InputStream;
+import java.util.function.Predicate;
import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers;
@@ -65,4 +66,12 @@ public interface FileMetadataCache {
MemoryBufferOrBuffers putFileMetadata(Object fileKey, int length,
InputStream is, CacheTag tag, AtomicBoolean isStopped) throws IOException;
+
+ /**
+ * Iterates through the file entries of this cache and for those that match the given predicate (aka have a matching
+ * CacheTag) will have their buffers marked for (a later) proactive eviction.
+ * @param predicate - matching the predicate indicates eligibility for proactive eviction
+ * @return number of bytes marked in the buffers eligible for eviction
+ */
+ long markBuffersForProactiveEviction(Predicate<CacheTag> predicate);
}