Big change to the way unsafe cache values are handled. Finlalize is no longer needed.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/52645dee Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/52645dee Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/52645dee Branch: refs/heads/master Commit: 52645deea393e4cd980b1510dc2fdff1450c89c4 Parents: ac447e2 Author: Aaron McCurry <[email protected]> Authored: Fri Oct 24 10:12:50 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Fri Oct 24 10:12:50 2014 -0400 ---------------------------------------------------------------------- .../blur/thrift/ThriftBlurShardServer.java | 8 +- .../blur/lucene/search/PrimeDocCache.java | 1 + .../blur/store/BlockCacheDirectoryFactory.java | 3 +- .../store/BlockCacheDirectoryFactoryV1.java | 5 + .../store/BlockCacheDirectoryFactoryV2.java | 5 + .../blur/store/blockcache_v2/BaseCache.java | 11 +- .../store/blockcache_v2/CacheDirectory.java | 6 +- .../store/blockcache_v2/CacheIndexInput.java | 160 +++++++++++++------ .../store/blockcache_v2/CacheIndexOutput.java | 2 - .../blur/store/blockcache_v2/CacheValue.java | 34 ++-- .../store/blockcache_v2/EvictionException.java | 23 +++ .../cachevalue/BaseCacheValue.java | 40 ++--- .../cachevalue/ByteArrayCacheValue.java | 1 + .../cachevalue/DetachableCacheValue.java | 57 +++---- .../cachevalue/UnsafeCacheValue.java | 23 ++- .../blockcache_v2/CacheIndexInputTest.java | 6 +- .../cachevalue/ByteArrayCacheValueTest.java | 5 +- .../cachevalue/UnsafeCacheValueTest.java | 9 +- 18 files changed, 248 insertions(+), 151 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java index faf3cf0..56dd688 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java @@ -161,7 +161,7 @@ public class ThriftBlurShardServer extends ThriftServer { } } - BlockCacheDirectoryFactory blockCacheDirectoryFactory; + final BlockCacheDirectoryFactory blockCacheDirectoryFactory; // Alternate BlockCacheDirectoryFactory support currently disabled in 0.2.0, // look for it in 0.2.1 String blockCacheVersion = configuration.get(BLUR_SHARD_BLOCK_CACHE_VERSION, "v2"); @@ -205,7 +205,7 @@ public class ThriftBlurShardServer extends ThriftServer { int internalSearchThreads = configuration.getInt(BLUR_SHARD_INTERNAL_SEARCH_THREAD_COUNT, 16); final DistributedIndexServer indexServer = new DistributedIndexServer(config, zooKeeper, clusterStatus, filterCache, blockCacheDirectoryFactory, distributedLayoutFactory, cluster, nodeName, safeModeDelay, - shardOpenerThreadCount, maxMergeThreads,internalSearchThreads, minimumNumberOfNodesBeforeExitingSafeMode); + shardOpenerThreadCount, maxMergeThreads, internalSearchThreads, minimumNumberOfNodesBeforeExitingSafeMode); BooleanQuery.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024)); @@ -301,8 +301,8 @@ public class ThriftBlurShardServer extends ThriftServer { @Override public void shutdown() { ThreadWatcher threadWatcher = ThreadWatcher.instance(); - quietClose(commandManager, traceStorage, refresher, server, shardServer, indexManager, indexServer, - threadWatcher, clusterStatus, zooKeeper, httpServer); + quietClose(blockCacheDirectoryFactory, commandManager, traceStorage, refresher, server, shardServer, + indexManager, indexServer, threadWatcher, clusterStatus, zooKeeper, httpServer); } }; server.setShutdown(shutdown); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-query/src/main/java/org/apache/blur/lucene/search/PrimeDocCache.java ---------------------------------------------------------------------- diff --git a/blur-query/src/main/java/org/apache/blur/lucene/search/PrimeDocCache.java b/blur-query/src/main/java/org/apache/blur/lucene/search/PrimeDocCache.java index 0c31359..d78b8b7 100644 --- a/blur-query/src/main/java/org/apache/blur/lucene/search/PrimeDocCache.java +++ b/blur-query/src/main/java/org/apache/blur/lucene/search/PrimeDocCache.java @@ -75,6 +75,7 @@ public class PrimeDocCache { if (count == docFreq) { primeDocMap.put(key, bs); } else { + // @TODO deal with deletes correctly... docFreq does not reflect deletes LOG.info("PrimeDoc for reader [{0}] not stored, because count [{1}] and freq [{2}] do not match.", reader, count, docFreq); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactory.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactory.java b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactory.java index a02d4cb..4473778 100644 --- a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactory.java +++ b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactory.java @@ -16,12 +16,13 @@ */ package org.apache.blur.store; +import java.io.Closeable; import java.io.IOException; import java.util.Set; import org.apache.lucene.store.Directory; -public abstract class BlockCacheDirectoryFactory { +public abstract class BlockCacheDirectoryFactory implements Closeable { public abstract Directory newDirectory(String table, String shard, Directory directory, Set<String> blockCacheFileTypes) throws IOException; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV1.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV1.java b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV1.java index 254b765..a622649 100644 --- a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV1.java +++ b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV1.java @@ -91,4 +91,9 @@ public class BlockCacheDirectoryFactoryV1 extends BlockCacheDirectoryFactory { return slabCount; } + @Override + public void close() throws IOException { + + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java index 759b12c..d1c279f 100644 --- a/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java +++ b/blur-store/src/main/java/org/apache/blur/store/BlockCacheDirectoryFactoryV2.java @@ -182,4 +182,9 @@ public class BlockCacheDirectoryFactoryV2 extends BlockCacheDirectoryFactory { return fileName.substring(indexOf + 1); } + @Override + public void close() throws IOException { + _cache.close(); + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/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 4805d40..0be0378 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 @@ -29,6 +29,7 @@ import static org.apache.blur.metrics.MetricsConstants.SIZE; import java.io.Closeable; import java.io.FileNotFoundException; import java.io.IOException; +import java.util.Collection; import java.util.HashSet; import java.util.Iterator; import java.util.Map; @@ -160,11 +161,19 @@ public class BaseCache extends Cache implements Closeable { @Override public void close() throws IOException { _running.set(false); - _cacheMap.clear(); + closeCachMap(); _oldFileDaemonThread.interrupt(); _cacheValueBufferPool.close(); } + private void closeCachMap() { + Collection<CacheValue> values = _cacheMap.values(); + for (CacheValue cacheValue : values) { + cacheValue.release(); + } + _cacheMap.clear(); + } + @Override public boolean shouldBeQuiet(CacheDirectory directory, String fileName) { return _quiet.shouldBeQuiet(directory, fileName); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheDirectory.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheDirectory.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheDirectory.java index 92af756..04f68d2 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheDirectory.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheDirectory.java @@ -61,11 +61,6 @@ public class CacheDirectory extends Directory implements DirectoryDecorator, Las return _table; } - @Override - protected void finalize() throws Throwable { - _cache.releaseDirectory(getDirectoryName()); - } - public IndexInput openInput(String name, IOContext context) throws IOException { IndexInput indexInput = _internal.openInput(name, context); if (_cache.cacheFileForReading(this, name, context) || isCachableFile(name)) { @@ -126,6 +121,7 @@ public class CacheDirectory extends Directory implements DirectoryDecorator, Las } public void close() throws IOException { + _cache.releaseDirectory(getDirectoryName()); _internal.close(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/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 3fdb5b3..7d9f933 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 @@ -20,6 +20,7 @@ package org.apache.blur.store.blockcache_v2; import java.io.EOFException; import java.io.IOException; +import org.apache.blur.store.blockcache_v2.cachevalue.ByteArrayCacheValue; import org.apache.blur.store.buffer.BufferStore; import org.apache.blur.store.buffer.Store; import org.apache.lucene.store.AlreadyClosedException; @@ -39,7 +40,7 @@ public class CacheIndexInput extends IndexInput { private IndexInput _indexInput; private CacheKey _key = new CacheKey(); private CacheValue _cacheValue; - private CacheValue _cacheValueQuietRef; + private CacheValue _cacheValueQuietRefCannotBeReleased; private long _position; private int _blockPosition; @@ -67,23 +68,44 @@ public class CacheIndexInput extends IndexInput { @Override public int readVInt() throws IOException { if (isCacheValueValid() && remaining() >= 5) { - byte b = readByteFromCache(); + byte b; + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } if (b >= 0) return b; int i = b & 0x7F; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7F) << 7; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7F) << 14; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7F) << 21; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } // Warning: the next ands use 0x0F / 0xF0 - beware copy/paste errors: i |= (b & 0x0F) << 28; if ((b & 0xF0) == 0) @@ -94,7 +116,7 @@ public class CacheIndexInput extends IndexInput { } private boolean isCacheValueValid() { - if (_cacheValue != null) { + if (_cacheValue != null && !_cacheValue.isEvicted()) { return true; } return false; @@ -103,39 +125,76 @@ public class CacheIndexInput extends IndexInput { @Override public long readVLong() throws IOException { if (isCacheValueValid() && remaining() >= 9) { - byte b = readByteFromCache(); + byte b; + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } if (b >= 0) return b; long i = b & 0x7FL; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7FL) << 7; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7FL) << 14; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7FL) << 21; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7FL) << 28; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7FL) << 35; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e2) { + b = readByte(); + } i |= (b & 0x7FL) << 42; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e1) { + b = readByte(); + } i |= (b & 0x7FL) << 49; if (b >= 0) return i; - b = readByteFromCache(); + try { + b = readByteFromCache(); + } catch (EvictionException e) { + b = readByte(); + } i |= (b & 0x7FL) << 56; if (b >= 0) return i; @@ -147,8 +206,14 @@ public class CacheIndexInput extends IndexInput { @Override public byte readByte() throws IOException { ensureOpen(); + byte b; tryToFill(); - byte b = _cacheValue.read(_blockPosition); + try { + b = _cacheValue.read(_blockPosition); + } catch (EvictionException e) { + releaseCache(); + return readByte(); + } _position++; _blockPosition++; return b; @@ -161,7 +226,13 @@ public class CacheIndexInput extends IndexInput { tryToFill(); int remaining = remaining(); int length = Math.min(len, remaining); - _cacheValue.read(_blockPosition, b, offset, length); + try { + _cacheValue.read(_blockPosition, b, offset, length); + } catch (EvictionException e) { + releaseCache(); + readBytes(b, offset, len); + return; + } offset += length; len -= length; _position += length; @@ -173,7 +244,12 @@ public class CacheIndexInput extends IndexInput { public short readShort() throws IOException { ensureOpen(); if (isCacheValueValid() && remaining() >= 2) { - short s = _cacheValue.readShort(_blockPosition); + short s; + try { + s = _cacheValue.readShort(_blockPosition); + } catch (EvictionException e) { + return super.readShort(); + } _blockPosition += 2; _position += 2; return s; @@ -185,7 +261,12 @@ public class CacheIndexInput extends IndexInput { public int readInt() throws IOException { ensureOpen(); if (isCacheValueValid() && remaining() >= 4) { - int i = _cacheValue.readInt(_blockPosition); + int i; + try { + i = _cacheValue.readInt(_blockPosition); + } catch (EvictionException e) { + return super.readInt(); + } _blockPosition += 4; _position += 4; return i; @@ -197,7 +278,12 @@ public class CacheIndexInput extends IndexInput { public long readLong() throws IOException { ensureOpen(); if (isCacheValueValid() && remaining() >= 8) { - long l = _cacheValue.readLong(_blockPosition); + long l; + try { + l = _cacheValue.readLong(_blockPosition); + } catch (EvictionException e) { + return super.readLong(); + } _blockPosition += 8; _position += 8; return l; @@ -212,11 +298,6 @@ public class CacheIndexInput extends IndexInput { _indexInput.close(); releaseCache(); } - if (_cacheValueQuietRef != null) { - CacheValue ref = _cacheValueQuietRef; - _cacheValueQuietRef = null; - ref.release(); - } } @Override @@ -271,19 +352,11 @@ public class CacheIndexInput extends IndexInput { clone._key = _key.clone(); clone._indexInput = _indexInput.clone(); clone._quiet = _cache.shouldBeQuiet(_directory, _fileName); - if (clone._cacheValue != null) { - clone._cacheValue.incRef(); - } - clone._cacheValueQuietRef = null; + clone._cacheValueQuietRefCannotBeReleased = null; return clone; } - @Override - protected void finalize() throws Throwable { - close(); - } - - private byte readByteFromCache() { + private byte readByteFromCache() throws EvictionException { byte b = _cacheValue.read(_blockPosition); _position++; _blockPosition++; @@ -306,7 +379,6 @@ public class CacheIndexInput extends IndexInput { private void releaseCache() { if (_cacheValue != null) { - _cacheValue.decRef(); _cacheValue = null; } } @@ -315,11 +387,12 @@ public class CacheIndexInput extends IndexInput { _key.setBlockId(getBlockId()); _cacheValue = _cache.getQuietly(_key); if (_cacheValue == null) { - if (_cacheValueQuietRef == null) { - _cacheValueQuietRef = _cache.newInstance(_directory, _fileName); + if (_cacheValueQuietRefCannotBeReleased == null) { + // @TODO this could be improved. + int cacheBlockSize = _cache.getCacheBlockSize(_directory, _fileName); + _cacheValueQuietRefCannotBeReleased = new ByteArrayCacheValue(cacheBlockSize); } - _cacheValue = _cacheValueQuietRef; - _cacheValue.incRef(); + _cacheValue = _cacheValueQuietRefCannotBeReleased; long filePosition = getFilePosition(); _indexInput.seek(filePosition); byte[] buffer = _store.takeBuffer(_bufferSize); @@ -333,8 +406,6 @@ public class CacheIndexInput extends IndexInput { cachePosition += length; } _store.putBuffer(buffer); - } else { - _cacheValue.incRef(); } _blockPosition = getBlockPosition(); } @@ -344,7 +415,6 @@ public class CacheIndexInput extends IndexInput { _cacheValue = _cache.get(_key); if (_cacheValue == null) { _cacheValue = _cache.newInstance(_directory, _fileName); - _cacheValue.incRef(); long filePosition = getFilePosition(); _indexInput.seek(filePosition); byte[] buffer = _store.takeBuffer(_bufferSize); @@ -359,8 +429,6 @@ public class CacheIndexInput extends IndexInput { } _store.putBuffer(buffer); _cache.put(_key.clone(), _cacheValue); - } else { - _cacheValue.incRef(); } _blockPosition = getBlockPosition(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java index caad489..120dc81 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/CacheIndexOutput.java @@ -97,12 +97,10 @@ public class CacheIndexOutput extends IndexOutput { } if (!_shouldBeQuiet) { CacheValue cacheValue = _cache.newInstance(_directory, _fileName); - cacheValue.incRef(); cacheValue.write(0, _buffer, 0, length); long blockId = (_position - length) / _cacheBlockSize; cacheValue = cacheValue.trim(length); _cache.put(new CacheKey(_fileId, blockId), cacheValue); - cacheValue.decRef(); } _bufferPosition = 0; writeBufferToOutputStream(length); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/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 997f24d..b9fa8f6 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,13 +20,14 @@ package org.apache.blur.store.blockcache_v2; import javax.swing.text.Position; public interface CacheValue { - + /** * Detach from the base cache. + * * @return old cache value. */ CacheValue detachFromCache(); - + /** * The length of the data in this block. * @@ -60,7 +61,7 @@ public interface CacheValue { * @param length * the length of data to read. */ - void read(int position, byte[] buf, int offset, int length); + void read(int position, byte[] buf, int offset, int length) throws EvictionException; /** * Reads a byte from the given position. @@ -69,22 +70,7 @@ public interface CacheValue { * the position. * @return the byte. */ - byte read(int position); - -// /** -// * Increments the reference. -// */ -// void incRef(); -// -// /** -// * Decrements the reference. -// */ -// void decRef(); -// -// /** -// * Gets the reference count. -// */ -// long refCount(); + byte read(int position) throws EvictionException; /** * Releases any underlying resources. @@ -98,7 +84,7 @@ public interface CacheValue { * the {@link Position} to read from. * @return the short. */ - short readShort(int position); + short readShort(int position) throws EvictionException; /** * Reads a int from the given position. @@ -107,7 +93,7 @@ public interface CacheValue { * the {@link Position} to read from. * @return the int. */ - int readInt(int position); + int readInt(int position) throws EvictionException; /** * Reads a long from the given position. @@ -116,7 +102,7 @@ public interface CacheValue { * the {@link Position} to read from. * @return the long. */ - long readLong(int position); + long readLong(int position) throws EvictionException; /** * This method <i>may</i> trim the existing {@link CacheValue} and produce @@ -130,8 +116,6 @@ public interface CacheValue { */ CacheValue trim(int length); - void decRef(); + boolean isEvicted(); - void incRef(); - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/EvictionException.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/EvictionException.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/EvictionException.java new file mode 100644 index 0000000..4870528 --- /dev/null +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/EvictionException.java @@ -0,0 +1,23 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.blur.store.blockcache_v2; + +public class EvictionException extends Exception { + + private static final long serialVersionUID = -5956627469112676201L; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/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 dc45eab..0cef01a 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 @@ -25,19 +25,15 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.blur.metrics.AtomicLongGauge; import org.apache.blur.store.blockcache_v2.CacheValue; +import org.apache.blur.store.blockcache_v2.EvictionException; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.MetricName; -@SuppressWarnings("serial") public abstract class BaseCacheValue implements CacheValue { private static final AtomicLong _neededFinalizedCall = new AtomicLong(); - public static class Evicted extends RuntimeException { - - } - private final int _length; protected volatile boolean _released = false; protected volatile boolean _evicted = false; @@ -58,21 +54,21 @@ public abstract class BaseCacheValue 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() { + private void checkForEviction() throws EvictionException { if (_evicted) { - throw new Evicted(); + throw new EvictionException(); } } @Override - public void read(int position, byte[] buf, int offset, int length) { + public void read(int position, byte[] buf, int offset, int length) throws EvictionException { checkForEviction(); if (position + length > _length) { throw new ArrayIndexOutOfBoundsException(position + length); @@ -81,7 +77,7 @@ public abstract class BaseCacheValue implements CacheValue { } @Override - public byte read(int position) { + public byte read(int position) throws EvictionException { checkForEviction(); if (position >= _length) { throw new ArrayIndexOutOfBoundsException(position); @@ -90,7 +86,7 @@ public abstract class BaseCacheValue implements CacheValue { } @Override - public short readShort(int position) { + public short readShort(int position) throws EvictionException { checkForEviction(); if (position + 2 > _length) { throw new ArrayIndexOutOfBoundsException(position + 2); @@ -103,7 +99,7 @@ public abstract class BaseCacheValue implements CacheValue { } @Override - public int readInt(int position) { + public int readInt(int position) throws EvictionException { checkForEviction(); if (position + 4 > _length) { throw new ArrayIndexOutOfBoundsException(position + 4); @@ -117,7 +113,7 @@ public abstract class BaseCacheValue implements CacheValue { } @Override - public long readLong(int position) { + public long readLong(int position) throws EvictionException { checkForEviction(); if (position + 8 > _length) { throw new ArrayIndexOutOfBoundsException(position + 4); @@ -136,15 +132,6 @@ public abstract class BaseCacheValue implements CacheValue { protected abstract void readInternal(int position, byte[] buf, int offset, int length); @Override - protected void finalize() throws Throwable { - // @TODO this may not be needed. - if (!_released) { - release(); - _neededFinalizedCall.incrementAndGet(); - } - } - - @Override public CacheValue trim(int length) { return this; } @@ -155,13 +142,8 @@ public abstract class BaseCacheValue implements CacheValue { } @Override - public void decRef() { - - } - - @Override - public void incRef() { - + public final boolean isEvicted() { + return _evicted; } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValue.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValue.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValue.java index e36f041..6b43149 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValue.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValue.java @@ -57,4 +57,5 @@ public class ByteArrayCacheValue extends BaseCacheValue { System.arraycopy(_buffer, 0, cacheValue._buffer, 0, length); return cacheValue; } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/DetachableCacheValue.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/DetachableCacheValue.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/DetachableCacheValue.java index 959e747..25c994c 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/DetachableCacheValue.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/DetachableCacheValue.java @@ -21,16 +21,15 @@ import static org.apache.blur.metrics.MetricsConstants.DETACHES; import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import org.apache.blur.store.blockcache_v2.CacheValue; +import org.apache.blur.store.blockcache_v2.EvictionException; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Meter; import com.yammer.metrics.core.MetricName; -@SuppressWarnings("serial") -public class DetachableCacheValue extends AtomicInteger implements CacheValue { +public class DetachableCacheValue implements CacheValue { private static final Meter _detaches; @@ -39,6 +38,7 @@ public class DetachableCacheValue extends AtomicInteger implements CacheValue { } private volatile CacheValue _baseCacheValue; + private volatile boolean _evicted; public DetachableCacheValue(CacheValue cacheValue) { _baseCacheValue = cacheValue; @@ -46,23 +46,16 @@ public class DetachableCacheValue extends AtomicInteger implements CacheValue { @Override public CacheValue detachFromCache() { + _evicted = true; if (_baseCacheValue instanceof ByteArrayCacheValue) { // already detached return null; } else if (_baseCacheValue instanceof UnsafeCacheValue) { final CacheValue result = _baseCacheValue; - if (get() == 0) { - // No one is using this so don't copy - // NULL out reference so just in case there can't be a seg fault. - _baseCacheValue = null; - } else { - // Copy data, because someone might access at some point - _detaches.mark(); - int length = _baseCacheValue.length(); - ByteArrayCacheValue byteArrayCacheValue = new ByteArrayCacheValue(length); - _baseCacheValue.read(0, byteArrayCacheValue._buffer, 0, length); - _baseCacheValue = byteArrayCacheValue; - } + // No one is using this so don't copy + // NULL out reference so just in case there can't be a seg fault. + _baseCacheValue = null; + _detaches.mark(); return result; } else { throw new RuntimeException("Unsupported type of [" + _baseCacheValue + "]"); @@ -80,32 +73,45 @@ public class DetachableCacheValue extends AtomicInteger implements CacheValue { } @Override - public void read(int position, byte[] buf, int offset, int length) { + public void read(int position, byte[] buf, int offset, int length) throws EvictionException { + checkEviction(); _baseCacheValue.read(position, buf, offset, length); } + private void checkEviction() throws EvictionException { + if (_evicted) { + throw new EvictionException(); + } + } + @Override - public byte read(int position) { + public byte read(int position) throws EvictionException { + checkEviction(); return _baseCacheValue.read(position); } @Override public void release() { - _baseCacheValue.release(); + if (_baseCacheValue != null) { + _baseCacheValue.release(); + } } @Override - public short readShort(int position) { + public short readShort(int position) throws EvictionException { + checkEviction(); return _baseCacheValue.readShort(position); } @Override - public int readInt(int position) { + public int readInt(int position) throws EvictionException { + checkEviction(); return _baseCacheValue.readInt(position); } @Override - public long readLong(int position) { + public long readLong(int position) throws EvictionException { + checkEviction(); return _baseCacheValue.readLong(position); } @@ -115,13 +121,8 @@ public class DetachableCacheValue extends AtomicInteger implements CacheValue { } @Override - public void decRef() { - decrementAndGet(); - } - - @Override - public void incRef() { - incrementAndGet(); + public boolean isEvicted() { + return _evicted; } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValue.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValue.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValue.java index ddff81f..6a37916 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValue.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValue.java @@ -17,6 +17,7 @@ */ package org.apache.blur.store.blockcache_v2.cachevalue; +import static org.apache.blur.metrics.MetricsConstants.CACHE_VALUE_FINALIZE; import static org.apache.blur.metrics.MetricsConstants.JVM; import static org.apache.blur.metrics.MetricsConstants.OFF_HEAP_MEMORY; import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; @@ -33,6 +34,13 @@ import com.yammer.metrics.core.MetricName; public class UnsafeCacheValue extends BaseCacheValue { + private static final AtomicLong _neededFinalizedCall = new AtomicLong(); + + static { + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, JVM, CACHE_VALUE_FINALIZE), new AtomicLongGauge( + _neededFinalizedCall)); + } + private static final Unsafe _unsafe; private static final AtomicLong _offHeapMemorySize = new AtomicLong(); @@ -88,9 +96,18 @@ public class UnsafeCacheValue extends BaseCacheValue { _unsafe.freeMemory(_address); _released = true; _offHeapMemorySize.addAndGet(0 - _capacity); -// } else { - // @TODO this is here to debug against double releases. -// new Throwable().printStackTrace(); } } + + // This is commented out normally. Add code when debugging memory related + // issues. + // @Override + // protected void finalize() throws Throwable { + // if (!_released) { + // new Throwable().printStackTrace(); + // System.exit(1); + // release(); + // _neededFinalizedCall.incrementAndGet(); + // } + // } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java index 36adfcc..9cbab58 100644 --- a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java +++ b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/CacheIndexInputTest.java @@ -23,6 +23,7 @@ import static org.junit.Assert.assertEquals; import java.io.IOException; import java.util.Random; +import org.apache.blur.store.blockcache_v2.cachevalue.ByteArrayCacheValue; import org.apache.blur.store.blockcache_v2.cachevalue.UnsafeCacheValue; import org.apache.blur.store.buffer.BufferStore; import org.apache.lucene.store.IOContext; @@ -193,7 +194,7 @@ public class CacheIndexInputTest { @Override public CacheValue newInstance(CacheDirectory directory, String fileName, int cacheBlockSize) { - return new UnsafeCacheValue(cacheBlockSize); + return new ByteArrayCacheValue(cacheBlockSize); } @Override @@ -258,11 +259,10 @@ public class CacheIndexInputTest { @Override public void fileClosedForWriting(CacheDirectory directory, String fileName, long fileId) throws IOException { - + } }; return cacheFactory; } - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValueTest.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValueTest.java b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValueTest.java index 8d33ece..ce799ce 100644 --- a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValueTest.java +++ b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/ByteArrayCacheValueTest.java @@ -19,12 +19,13 @@ package org.apache.blur.store.blockcache_v2.cachevalue; import static org.junit.Assert.*; +import org.apache.blur.store.blockcache_v2.EvictionException; import org.junit.Test; public class ByteArrayCacheValueTest { @Test - public void test1() { + public void test1() throws EvictionException { ByteArrayCacheValue value = new ByteArrayCacheValue(10); byte[] buf = "hello world".getBytes(); value.write(0, buf, 0, 10); @@ -56,7 +57,7 @@ public class ByteArrayCacheValueTest { } @Test - public void test4() { + public void test4() throws EvictionException { ByteArrayCacheValue value = new ByteArrayCacheValue(10); byte[] buf = "hello world".getBytes(); value.write(8, buf, 0, 2); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/52645dee/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValueTest.java ---------------------------------------------------------------------- diff --git a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValueTest.java b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValueTest.java index 37f5926..62095f8 100644 --- a/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValueTest.java +++ b/blur-store/src/test/java/org/apache/blur/store/blockcache_v2/cachevalue/UnsafeCacheValueTest.java @@ -19,18 +19,20 @@ package org.apache.blur.store.blockcache_v2.cachevalue; import static org.junit.Assert.*; +import org.apache.blur.store.blockcache_v2.EvictionException; import org.junit.Test; public class UnsafeCacheValueTest { @Test - public void test1() { + public void test1() throws EvictionException { UnsafeCacheValue value = new UnsafeCacheValue(10); byte[] buf = "hello world".getBytes(); value.write(0, buf, 0, 10); byte[] buf2 = new byte[10]; value.read(0, buf2, 0, 10); assertArrayEquals("hello worl".getBytes(), buf2); + value.release(); } @Test @@ -42,6 +44,7 @@ public class UnsafeCacheValueTest { fail(); } catch (ArrayIndexOutOfBoundsException e) { } + value.release(); } @Test @@ -53,16 +56,18 @@ public class UnsafeCacheValueTest { fail(); } catch (ArrayIndexOutOfBoundsException e) { } + value.release(); } @Test - public void test4() { + public void test4() throws EvictionException { UnsafeCacheValue value = new UnsafeCacheValue(10); byte[] buf = "hello world".getBytes(); value.write(8, buf, 0, 2); assertEquals('h', value.read(8)); assertEquals('e', value.read(9)); + value.release(); } }
