Changing the behavior of the block cache. When a cachevalue reference is held past eviction the eviction flag is set and a release is forced a few seconds later.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/5c5a38f1 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/5c5a38f1 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/5c5a38f1 Branch: refs/heads/apache-blur-0.2 Commit: 5c5a38f150fc0793c744a902be3af629da3384e9 Parents: 81df87e Author: Aaron McCurry <[email protected]> Authored: Wed Nov 20 10:25:41 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Wed Nov 20 10:25:41 2013 -0500 ---------------------------------------------------------------------- .../blur/store/blockcache_v2/BaseCache.java | 17 +++++++---- .../store/blockcache_v2/CacheIndexInput.java | 23 +++++++++------ .../blur/store/blockcache_v2/CacheValue.java | 11 +++++++ .../cachevalue/BaseCacheValue.java | 31 ++++++++++++++++++-- .../blur/store/BaseDirectoryTestSuite.java | 2 +- 5 files changed, 67 insertions(+), 17 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5c5a38f1/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java index ed95898..2e25ad8 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/BaseCache.java @@ -69,6 +69,7 @@ public class BaseCache extends Cache implements Closeable { @Override public void onEviction(CacheKey key, CacheValue value) { _evictions.mark(); + value.evict(); addToReleaseQueue(key, value); } } @@ -118,7 +119,7 @@ public class BaseCache extends Cache implements Closeable { private final Thread _oldCacheValueDaemonThread; private final AtomicBoolean _running = new AtomicBoolean(true); private final BlockingQueue<ReleaseEntry> _releaseQueue; - private final long _warningTimeForEntryCleanup = TimeUnit.MINUTES.toMillis(60); + private final long _warningTimeForEntryCleanup = TimeUnit.SECONDS.toMillis(10); public BaseCache(long totalNumberOfBytes, Size fileBufferSize, Size cacheBlockSize, FileNameFilter readFilter, FileNameFilter writeFilter, Quiet quiet, STORE store) { @@ -194,15 +195,20 @@ public class BaseCache extends Cache implements Closeable { entriesToCleanup.remove(fileId); CacheValue value = entry._value; + boolean release = false; if (value.refCount() == 0) { + release = true; + } else if (entry.hasLivedToLong(_warningTimeForEntryCleanup)) { + FileIdKey fileIdKey = _oldFileNameIdMap.get(fileId); + LOG.warn("CacheValue has not been released [{0}] for [{1}] for over [{2} ms] forcing release.", entry, + fileIdKey, _warningTimeForEntryCleanup); + release = true; + } + if (release) { value.release(); iterator.remove(); long capacity = _cacheMap.capacity(); _cacheMap.setCapacity(capacity + value.size()); - } else if (entry.hasLivedToLong(_warningTimeForEntryCleanup)) { - FileIdKey fileIdKey = _oldFileNameIdMap.get(fileId); - LOG.warn("CacheValue has not been released [{0}] for [{1}] for over [{2} ms]", entry, fileIdKey, - _warningTimeForEntryCleanup); } } for (Long l : entriesToCleanup.keySet()) { @@ -218,6 +224,7 @@ public class BaseCache extends Cache implements Closeable { if (!validFileIds.contains(fileId)) { CacheValue remove = _cacheMap.remove(key); if (remove != null) { + remove.evict(); _removals.mark(); addToReleaseQueue(key, remove); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5c5a38f1/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java index 95d9689..5d02963 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexInput.java @@ -64,7 +64,7 @@ public class CacheIndexInput extends IndexInput { @Override public int readVInt() throws IOException { - if (_cacheValue != null && remaining() >= 5) { + if (isCacheValueValid() && remaining() >= 5) { byte b = readByteFromCache(); if (b >= 0) return b; @@ -91,9 +91,16 @@ public class CacheIndexInput extends IndexInput { return super.readVInt(); } + private boolean isCacheValueValid() { + if (_cacheValue != null && !_cacheValue.isEvicted()) { + return true; + } + return false; + } + @Override public long readVLong() throws IOException { - if (_cacheValue != null && remaining() >= 9) { + if (isCacheValueValid() && remaining() >= 9) { byte b = readByteFromCache(); if (b >= 0) return b; @@ -163,7 +170,7 @@ public class CacheIndexInput extends IndexInput { @Override public short readShort() throws IOException { ensureOpen(); - if (_cacheValue != null && remaining() >= 2) { + if (isCacheValueValid() && remaining() >= 2) { short s = _cacheValue.readShort(_blockPosition); _blockPosition += 2; _position += 2; @@ -175,7 +182,7 @@ public class CacheIndexInput extends IndexInput { @Override public int readInt() throws IOException { ensureOpen(); - if (_cacheValue != null && remaining() >= 4) { + if (isCacheValueValid() && remaining() >= 4) { int i = _cacheValue.readInt(_blockPosition); _blockPosition += 4; _position += 4; @@ -187,7 +194,7 @@ public class CacheIndexInput extends IndexInput { @Override public long readLong() throws IOException { ensureOpen(); - if (_cacheValue != null && remaining() >= 8) { + if (isCacheValueValid() && remaining() >= 8) { long l = _cacheValue.readLong(_blockPosition); _blockPosition += 8; _position += 8; @@ -249,7 +256,7 @@ public class CacheIndexInput extends IndexInput { CacheIndexInput clone = (CacheIndexInput) super.clone(); clone._key = _key.clone(); clone._indexInput = _indexInput.clone(); - if (clone._cacheValue != null) { + if (isCacheValueValid()) { clone._cacheValue.incRef(); } clone._quiet = _cache.shouldBeQuiet(_directory, _fileName); @@ -273,9 +280,7 @@ public class CacheIndexInput extends IndexInput { } private void tryToFill() throws IOException { - if (_cacheValue == null) { - fill(); - } else if (remaining() == 0) { + if (!isCacheValueValid() || remaining() == 0) { releaseCache(); fill(); } else { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5c5a38f1/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheValue.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheValue.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheValue.java index 68da92f..0b24c47 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheValue.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheValue.java @@ -20,6 +20,17 @@ package org.apache.blur.store.blockcache_v2; import javax.swing.text.Position; public interface CacheValue { + + /** + * Marks value as evicted. + */ + void evict(); + + /** + * Gets whether or not value is evicted. + * @return + */ + boolean isEvicted(); /** * The actual size of the the underlying resource. http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5c5a38f1/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/BaseCacheValue.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/BaseCacheValue.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/BaseCacheValue.java index 319e42d..cf58beb 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/BaseCacheValue.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/BaseCacheValue.java @@ -32,10 +32,15 @@ import com.yammer.metrics.core.MetricName; @SuppressWarnings("serial") public abstract class BaseCacheValue extends AtomicLong implements CacheValue { + private static final AtomicLong _neededFinalizedCall = new AtomicLong(); + + public static class Evicted extends RuntimeException { + + } + private final int _length; protected volatile boolean _released = false; - - private static final AtomicLong _neededFinalizedCall = new AtomicLong(); + protected volatile boolean _evicted = false; static { Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, JVM, CACHE_VALUE_FINALIZE), new AtomicLongGauge( @@ -53,14 +58,22 @@ public abstract class BaseCacheValue extends AtomicLong implements CacheValue { @Override public void write(int position, byte[] buf, int offset, int length) { + checkForEviction(); if (position + length > _length) { throw new ArrayIndexOutOfBoundsException(position + length); } writeInternal(position, buf, offset, length); } + private void checkForEviction() { + if (_evicted) { + throw new Evicted(); + } + } + @Override public void read(int position, byte[] buf, int offset, int length) { + checkForEviction(); if (position + length > _length) { throw new ArrayIndexOutOfBoundsException(position + length); } @@ -69,6 +82,7 @@ public abstract class BaseCacheValue extends AtomicLong implements CacheValue { @Override public byte read(int position) { + checkForEviction(); if (position >= _length) { throw new ArrayIndexOutOfBoundsException(position); } @@ -77,6 +91,7 @@ public abstract class BaseCacheValue extends AtomicLong implements CacheValue { @Override public short readShort(int position) { + checkForEviction(); if (position + 2 > _length) { throw new ArrayIndexOutOfBoundsException(position + 2); } @@ -89,6 +104,7 @@ public abstract class BaseCacheValue extends AtomicLong implements CacheValue { @Override public int readInt(int position) { + checkForEviction(); if (position + 4 > _length) { throw new ArrayIndexOutOfBoundsException(position + 4); } @@ -102,6 +118,7 @@ public abstract class BaseCacheValue extends AtomicLong implements CacheValue { @Override public long readLong(int position) { + checkForEviction(); if (position + 8 > _length) { throw new ArrayIndexOutOfBoundsException(position + 4); } @@ -141,4 +158,14 @@ public abstract class BaseCacheValue extends AtomicLong implements CacheValue { _neededFinalizedCall.incrementAndGet(); } } + + @Override + public void evict() { + _evicted = true; + } + + @Override + public boolean isEvicted() { + return _evicted; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/5c5a38f1/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java b/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java index a70a1fe..03a0358 100644 --- a/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java +++ b/blur-store/src/test/java/org/apache/blur/store/BaseDirectoryTestSuite.java @@ -72,7 +72,7 @@ public abstract class BaseDirectoryTestSuite { public void setUp() throws IOException { BufferStore.initNewBuffer(1024, 1024 * 128); BufferStore.initNewBuffer(8192, 8192 * 128); - + file = new File(TMPDIR, "hdfsdirectorytest"); fileControl = new File(TMPDIR, "hdfsdirectorytest-control"); rm(file);
