Updated Branches: refs/heads/apache-blur-0.2 57245569d -> 4a338ad00
Fixed BLUR-281, also made a bunch of small changes to make the block cache more stable. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/4a338ad0 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/4a338ad0 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/4a338ad0 Branch: refs/heads/apache-blur-0.2 Commit: 4a338ad000ff0fd7344335dfea52a19fdf04f269 Parents: 5724556 Author: Aaron McCurry <[email protected]> Authored: Sun Oct 20 20:17:00 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Sun Oct 20 20:17:00 2013 -0400 ---------------------------------------------------------------------- .../blur/manager/writer/IndexImporter.java | 30 ++++-- .../manager/writer/SharedMergeScheduler.java | 20 +++- .../org/apache/blur/utils/GCWatcherJdk7.java | 19 +++- .../blur/store/blockcache_v2/BaseCache.java | 107 +++++++++++++++++-- .../apache/blur/store/blockcache_v2/Cache.java | 3 +- .../store/blockcache_v2/CacheIndexInput.java | 2 +- .../cachevalue/UnsafeCacheValue.java | 14 +++ .../blockcache_v2/CacheIndexInputTest.java | 5 + .../apache/blur/metrics/MetricsConstants.java | 4 + 9 files changed, 181 insertions(+), 23 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java index cbf5e91..1d66f3a 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java @@ -17,6 +17,7 @@ package org.apache.blur.manager.writer; * limitations under the License. */ import java.io.Closeable; +import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -84,15 +85,28 @@ public class IndexImporter extends TimerTask implements Closeable { Configuration configuration = _shardContext.getTableContext().getConfiguration(); try { FileSystem fileSystem = path.getFileSystem(configuration); - SortedSet<FileStatus> listStatus = sort(fileSystem.listStatus(path, new PathFilter() { - @Override - public boolean accept(Path path) { - if (path != null && path.getName().endsWith(".commit")) { - return true; - } - return false; + SortedSet<FileStatus> listStatus; + while (true) { + try { + listStatus = sort(fileSystem.listStatus(path, new PathFilter() { + @Override + public boolean accept(Path path) { + if (path != null && path.getName().endsWith(".commit")) { + return true; + } + return false; + } + })); + break; + } catch (FileNotFoundException e) { + LOG.warn("File not found error, retrying."); } - })); + try { + Thread.sleep(100); + } catch (InterruptedException e) { + return; + } + } List<HdfsDirectory> indexesToImport = new ArrayList<HdfsDirectory>(); for (FileStatus fileStatus : listStatus) { Path file = fileStatus.getPath(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java b/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java index 6fe8205..5036f37 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/SharedMergeScheduler.java @@ -16,6 +16,9 @@ package org.apache.blur.manager.writer; * See the License for the specific language governing permissions and * limitations under the License. */ +import static org.apache.blur.metrics.MetricsConstants.LUCENE; +import static org.apache.blur.metrics.MetricsConstants.MERGE_THROUGHPUT_BYTES; +import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; import static org.apache.blur.utils.BlurConstants.SHARED_MERGE_SCHEDULER; import java.io.Closeable; @@ -23,6 +26,7 @@ import java.io.IOException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.blur.concurrent.Executors; @@ -32,20 +36,27 @@ import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.MergePolicy; import org.apache.lucene.index.MergeScheduler; +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.Meter; +import com.yammer.metrics.core.MetricName; + public class SharedMergeScheduler implements Runnable, Closeable { private static final Log LOG = LogFactory.getLog(SharedMergeScheduler.class); - private static final long ONE_SECOND = 1000; + private static final long ONE_SECOND = TimeUnit.SECONDS.toMillis(1); - private BlockingQueue<IndexWriter> _writers = new LinkedBlockingQueue<IndexWriter>(); - private AtomicBoolean _running = new AtomicBoolean(true); - private ExecutorService _service; + private final BlockingQueue<IndexWriter> _writers = new LinkedBlockingQueue<IndexWriter>(); + private final AtomicBoolean _running = new AtomicBoolean(true); + private final ExecutorService _service; + private final Meter _throughputBytes; public SharedMergeScheduler(int threads) { _service = Executors.newThreadPool(SHARED_MERGE_SCHEDULER, threads, false); for (int i = 0; i < threads; i++) { _service.submit(this); } + MetricName mergeThoughputBytes = new MetricName(ORG_APACHE_BLUR, LUCENE, MERGE_THROUGHPUT_BYTES); + _throughputBytes = Metrics.newMeter(mergeThoughputBytes, MERGE_THROUGHPUT_BYTES, TimeUnit.SECONDS); } private void mergeIndexWriter(IndexWriter writer) { @@ -128,6 +139,7 @@ public class SharedMergeScheduler implements Runnable, Closeable { } else { LOG.debug("Merge took [{0} s] to complete at rate of [{1} MB/s]", time, rate); } + _throughputBytes.mark(merge.totalBytesSize()); return true; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/blur-core/src/main/java/org/apache/blur/utils/GCWatcherJdk7.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/utils/GCWatcherJdk7.java b/blur-core/src/main/java/org/apache/blur/utils/GCWatcherJdk7.java index b9de5a4..44dfb18 100644 --- a/blur-core/src/main/java/org/apache/blur/utils/GCWatcherJdk7.java +++ b/blur-core/src/main/java/org/apache/blur/utils/GCWatcherJdk7.java @@ -16,6 +16,8 @@ */ package org.apache.blur.utils; +import static org.apache.blur.metrics.MetricsConstants.*; + import java.lang.management.GarbageCollectorMXBean; import java.lang.management.ManagementFactory; import java.lang.management.MemoryMXBean; @@ -25,6 +27,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.TimeUnit; import javax.management.Notification; import javax.management.NotificationFilter; @@ -34,19 +37,29 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import com.sun.management.GcInfo; +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.Meter; +import com.yammer.metrics.core.MetricName; +import com.yammer.metrics.core.Timer; public class GCWatcherJdk7 { private static final Log LOG = LogFactory.getLog(GCWatcherJdk7.class); private static final String GET_LAST_GC_INFO = "getLastGcInfo"; + private static final long _1_SECOND = TimeUnit.SECONDS.toMillis(1); private static GCWatcherJdk7 _instance; private final MemoryMXBean _memoryMXBean; private final double _ratio; private final List<GCAction> _actions = new ArrayList<GCAction>(); + private final Timer _gcTimes; private GCWatcherJdk7(double ratio) { _memoryMXBean = ManagementFactory.getMemoryMXBean(); + + MetricName gcTimesName = new MetricName(ORG_APACHE_BLUR, JVM, GC_TIMES); + _gcTimes = Metrics.newTimer(gcTimesName, TimeUnit.MILLISECONDS, TimeUnit.SECONDS); + List<GarbageCollectorMXBean> garbageCollectorMXBeans = ManagementFactory.getGarbageCollectorMXBeans(); for (GarbageCollectorMXBean bean : garbageCollectorMXBeans) { NotificationListener listener = new NotificationListener() { @@ -61,8 +74,12 @@ public class GCWatcherJdk7 { long usedBefore = getTotal(usageBeforeGc); long usedAfter = getTotal(usageAfterGc); long totalTime = endTime - startTime; + long totalSize = usedBefore - usedAfter; + if (totalTime >= _1_SECOND) { - LOG.info("GC event totalTime spent in GC [{0} ms] collected [{1} bytes]", totalTime, (usedBefore - usedAfter)); + LOG.info("GC event totalTime spent in GC [{0} ms] collected [{1} bytes]", totalTime, totalSize); + } + _gcTimes.update(totalTime, TimeUnit.MILLISECONDS); MemoryUsage heapMemoryUsage = _memoryMXBean.getHeapMemoryUsage(); long max = heapMemoryUsage.getMax(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/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 84f4b32..6d80797 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 @@ -17,12 +17,25 @@ */ package org.apache.blur.store.blockcache_v2; +import static org.apache.blur.metrics.MetricsConstants.CACHE; +import static org.apache.blur.metrics.MetricsConstants.ENTRIES; +import static org.apache.blur.metrics.MetricsConstants.EVICTION; +import static org.apache.blur.metrics.MetricsConstants.HIT; +import static org.apache.blur.metrics.MetricsConstants.MISS; +import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; +import static org.apache.blur.metrics.MetricsConstants.REMOVAL; +import static org.apache.blur.metrics.MetricsConstants.SIZE; + +import java.io.Closeable; import java.io.IOException; +import java.util.HashSet; import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import org.apache.blur.log.Log; @@ -34,10 +47,15 @@ import org.apache.lucene.store.IOContext; import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap; import com.googlecode.concurrentlinkedhashmap.EvictionListener; import com.googlecode.concurrentlinkedhashmap.Weigher; +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.Gauge; +import com.yammer.metrics.core.Meter; +import com.yammer.metrics.core.MetricName; -public class BaseCache extends Cache { +public class BaseCache extends Cache implements Closeable { private static final Log LOG = LogFactory.getLog(BaseCache.class); + private static final long _1_MINUTE = TimeUnit.MINUTES.toMillis(1); public enum STORE { ON_HEAP, OFF_HEAP @@ -66,6 +84,12 @@ public class BaseCache extends Cache { private final Map<FileIdKey, Long> _fileNameToId = new ConcurrentHashMap<FileIdKey, Long>(); private final AtomicLong _fileId = new AtomicLong(); private final Quiet _quiet; + private final Meter _hits; + private final Meter _misses; + private final Meter _evictions; + private final Meter _removals; + private final Thread _daemonThread; + private final AtomicBoolean _running = new AtomicBoolean(true); public BaseCache(long totalNumberOfBytes, Size fileBufferSize, Size cacheBlockSize, FileNameFilter readFilter, FileNameFilter writeFilter, Quiet quiet, STORE store) { @@ -77,18 +101,73 @@ public class BaseCache extends Cache { _store = store; _cacheBlockSize = cacheBlockSize; _quiet = quiet; + _hits = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, CACHE, HIT), HIT, TimeUnit.SECONDS); + _misses = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, CACHE, MISS), MISS, TimeUnit.SECONDS); + _evictions = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, CACHE, EVICTION), EVICTION, TimeUnit.SECONDS); + _removals = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, CACHE, REMOVAL), REMOVAL, TimeUnit.SECONDS); + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, CACHE, ENTRIES), new Gauge<Long>() { + @Override + public Long value() { + return (long) _cacheMap.size(); + } + }); + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, CACHE, SIZE), new Gauge<Long>() { + @Override + public Long value() { + return _cacheMap.weightedSize(); + } + }); + _daemonThread = new Thread(new Runnable() { + @Override + public void run() { + while (_running.get()) { + cleanupOldFiles(); + try { + Thread.sleep(_1_MINUTE); + } catch (InterruptedException e) { + return; + } + } + } + }); + _daemonThread.setDaemon(true); + _daemonThread.setName("BaseCacheCleanup"); + _daemonThread.setPriority(Thread.MIN_PRIORITY); + _daemonThread.start(); + } + + protected void cleanupOldFiles() { + LOG.info("Cleanup old files from cache."); + Set<Long> validFileIds = new HashSet<Long>(_fileNameToId.values()); + for (CacheKey key : _cacheMap.keySet()) { + long fileId = key.getFileId(); + if (validFileIds.contains(fileId)) { + CacheValue remove = _cacheMap.remove(key); + if (remove != null) { + _removals.mark(); + } + } + } + } + + @Override + public void close() throws IOException { + _running.set(false); + _cacheMap.clear(); + _daemonThread.interrupt(); } private void addToReleaseQueue(CacheValue value) { if (value != null) { + _evictions.mark(); if (value.refCount() == 0) { value.release(); return; } - LOG.debug("CacheValue was not released [{0}]", value); + LOG.info("CacheValue was not released [{0}]", value); } } - + @Override public boolean shouldBeQuiet(CacheDirectory directory, String fileName) { return _quiet.shouldBeQuiet(directory.getDirectoryName(), fileName); @@ -151,17 +230,30 @@ public class BaseCache extends Cache { @Override public CacheValue get(CacheKey key) { - return _cacheMap.get(key); + CacheValue cacheValue = _cacheMap.get(key); + if (cacheValue == null) { + _misses.mark(); + } else { + _hits.mark(); + } + return cacheValue; } - + @Override public CacheValue getQuietly(CacheKey key) { - return _cacheMap.getQuietly(key); + CacheValue cacheValue = _cacheMap.getQuietly(key); + if (cacheValue != null) { + _hits.mark(); + } + return cacheValue; } @Override public void put(CacheKey key, CacheValue value) { - _cacheMap.put(key, value); + CacheValue cacheValue = _cacheMap.put(key, value); + if (cacheValue != null) { + _evictions.mark(); + } } @Override @@ -221,7 +313,6 @@ public class BaseCache extends Cache { return false; return true; } - } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java index bdc6b75..1dddd75 100644 --- a/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java +++ b/blur-store/src/main/java/org/apache/blur/store/blockcache_v2/Cache.java @@ -17,11 +17,12 @@ */ package org.apache.blur.store.blockcache_v2; +import java.io.Closeable; import java.io.IOException; import org.apache.lucene.store.IOContext; -public abstract class Cache { +public abstract class Cache implements Closeable { /** * Creates a new instance of CacheValue, the cache capacity should be used for http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/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 1f74edb..8967f99 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 @@ -225,8 +225,8 @@ public class CacheIndexInput extends IndexInput { cachePosition += length; } BufferStore.putBuffer(buffer); + _cache.put(_key.clone(), _cacheValue); } - _cache.put(_key.clone(), _cacheValue); _cacheValue.incRef(); _blockPosition = getBlockPosition(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/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 3ea3796..c954316 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,15 +17,26 @@ */ package org.apache.blur.store.blockcache_v2.cachevalue; +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; + import java.lang.reflect.Field; +import java.util.concurrent.atomic.AtomicLong; + +import org.apache.blur.metrics.AtomicLongGauge; import sun.misc.Unsafe; +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.MetricName; + @SuppressWarnings("serial") public class UnsafeCacheValue extends BaseCacheValue { private static final String JAVA_NIO_BITS = "java.nio.Bits"; private static final Unsafe _unsafe; + private static final AtomicLong _offHeapMemorySize = new AtomicLong(); static { try { @@ -36,6 +47,7 @@ public class UnsafeCacheValue extends BaseCacheValue { } catch (Exception e) { throw new RuntimeException(e); } + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, JVM, OFF_HEAP_MEMORY), new AtomicLongGauge(_offHeapMemorySize)); } private static final int BYTE_ARRAY_BASE_OFFSET = _unsafe.arrayBaseOffset(byte[].class); @@ -57,6 +69,7 @@ public class UnsafeCacheValue extends BaseCacheValue { super(length); _capacity = getCapacity(length); _address = _unsafe.allocateMemory(_capacity); + _offHeapMemorySize.addAndGet(_capacity); } private int getCapacity(int length) { @@ -87,6 +100,7 @@ public class UnsafeCacheValue extends BaseCacheValue { if (!_released) { _unsafe.freeMemory(_address); _released = true; + _offHeapMemorySize.addAndGet(0 - _capacity); } else { new Throwable().printStackTrace(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/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 4d488d4..77158bf 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 @@ -256,6 +256,11 @@ public class CacheIndexInputTest { return false; } + @Override + public void close() throws IOException { + + } + }; return cacheFactory; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4a338ad0/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java b/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java index bc008f5..71024be 100644 --- a/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java +++ b/blur-util/src/main/java/org/apache/blur/metrics/MetricsConstants.java @@ -39,6 +39,7 @@ public class MetricsConstants { public static final String HDFS = "HDFS"; public static final String LOCAL = "local"; public static final String HIT = "Hit"; + public static final String REMOVAL = "Removal"; public static final String MISS = "Miss"; public static final String CACHE = "Cache"; public static final String JVM = "JVM"; @@ -56,4 +57,7 @@ public class MetricsConstants { public static final String SIZE = "Size"; public static final String LOAD_AVERAGE = "Load Average"; public static final String SYSTEM = "System"; + public static final String MERGE_THROUGHPUT_BYTES = "Merge Throughput Bytes"; + public static final String GC_TIMES = "GC Times (ms)"; + public static final String OFF_HEAP_MEMORY = "Off Heap Memory"; }
