Many more trace updates.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/53c40727 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/53c40727 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/53c40727 Branch: refs/heads/apache-blur-0.2 Commit: 53c40727c2696052c40c2017ce312971cd760959 Parents: c19a3f1 Author: Aaron McCurry <[email protected]> Authored: Wed Nov 27 09:54:12 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Wed Nov 27 09:54:12 2013 -0500 ---------------------------------------------------------------------- .../indexserver/DefaultBlurIndexWarmup.java | 29 ++++++++++-- .../blur/manager/writer/BlurNRTIndex.java | 6 ++- .../blur/manager/writer/FieldBasedWarmer.java | 24 +++++----- .../apache/blur/lucene/warmup/IndexWarmup.java | 16 ++++--- .../main/java/org/apache/blur/trace/Trace.java | 17 +++++-- .../org/apache/blur/trace/TraceCollector.java | 14 +++--- .../java/org/apache/blur/trace/TracerImpl.java | 50 ++++++++++++++++---- .../apache/blur/zookeeper/ZooKeeperClient.java | 13 ++++- 8 files changed, 125 insertions(+), 44 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java index 0e96b8f..2320c16 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DefaultBlurIndexWarmup.java @@ -18,6 +18,8 @@ package org.apache.blur.manager.indexserver; */ import java.io.IOException; import java.lang.reflect.Field; +import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -32,6 +34,8 @@ import org.apache.blur.lucene.warmup.IndexTracerResult; import org.apache.blur.lucene.warmup.IndexWarmup; import org.apache.blur.manager.indexserver.DistributedIndexServer.ReleaseReader; import org.apache.blur.thrift.generated.TableDescriptor; +import org.apache.blur.utils.GCAction; +import org.apache.blur.utils.GCWatcher; import org.apache.lucene.index.AtomicReader; import org.apache.lucene.index.AtomicReaderContext; import org.apache.lucene.index.FilterDirectoryReader; @@ -41,22 +45,38 @@ import org.apache.lucene.index.SegmentReader; public class DefaultBlurIndexWarmup extends BlurIndexWarmup { + private static final Log LOG = LogFactory.getLog(DefaultBlurIndexWarmup.class); + + private Set<AtomicBoolean> _stops = Collections.synchronizedSet(new HashSet<AtomicBoolean>()); + public DefaultBlurIndexWarmup(long warmupBandwidthThrottleBytesPerSec) { super(warmupBandwidthThrottleBytesPerSec); + GCWatcher.registerAction(new GCAction() { + @Override + public void takeAction() throws Exception { + synchronized (_stops) { + for (AtomicBoolean stop : _stops) { + stop.set(true); + } + } + } + }); } - private static final Log LOG = LogFactory.getLog(DefaultBlurIndexWarmup.class); - @Override public void warmBlurIndex(final TableDescriptor table, final String shard, IndexReader reader, AtomicBoolean isClosed, ReleaseReader releaseReader, AtomicLong pauseWarmup) throws IOException { LOG.info("Running warmup for reader [{0}]", reader); + AtomicBoolean stop = new AtomicBoolean(); try { if (reader instanceof FilterDirectoryReader) { reader = getBase((FilterDirectoryReader) reader); } int maxSampleSize = 1000; - IndexWarmup indexWarmup = new IndexWarmup(isClosed, maxSampleSize, _warmupBandwidthThrottleBytesPerSec); + synchronized (_stops) { + _stops.add(stop); + } + IndexWarmup indexWarmup = new IndexWarmup(isClosed, stop, maxSampleSize, _warmupBandwidthThrottleBytesPerSec); String context = table.getName() + "/" + shard; Map<String, List<IndexTracerResult>> sampleIndex = indexWarmup.sampleIndex(reader, context); List<String> preCacheCols = table.getPreCacheCols(); @@ -66,6 +86,9 @@ public class DefaultBlurIndexWarmup extends BlurIndexWarmup { warm(reader, getFields(reader), indexWarmup, sampleIndex, context, isClosed, pauseWarmup); } } finally { + synchronized (_stops) { + _stops.remove(stop); + } releaseReader.release(); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java index b56b53b..76d7e58 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurNRTIndex.java @@ -107,7 +107,8 @@ public class BlurNRTIndex extends BlurIndex { conf.setWriteLockTimeout(TimeUnit.MINUTES.toMillis(5)); conf.setCodec(new Blur021Codec()); conf.setSimilarity(_tableContext.getSimilarity()); - conf.setMergedSegmentWarmer(new FieldBasedWarmer(shardContext, _isClosed)); + AtomicBoolean stop = new AtomicBoolean(); + conf.setMergedSegmentWarmer(new FieldBasedWarmer(shardContext, stop, _isClosed)); SnapshotDeletionPolicy sdp; if (snapshotsDirectoryExists()) { @@ -137,7 +138,8 @@ public class BlurNRTIndex extends BlurIndex { _searcherFactory = new SearcherFactory() { @Override public IndexSearcher newSearcher(IndexReader reader) throws IOException { -// return new IndexSearcherClosableNRT(reader, searchExecutor, _nrtManagerRef, _directory); + // return new IndexSearcherClosableNRT(reader, searchExecutor, + // _nrtManagerRef, _directory); return new IndexSearcherClosableNRT(reader, null, _nrtManagerRef, _directory); } }; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-core/src/main/java/org/apache/blur/manager/writer/FieldBasedWarmer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/FieldBasedWarmer.java b/blur-core/src/main/java/org/apache/blur/manager/writer/FieldBasedWarmer.java index 2cfc383..f77d4ef 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/FieldBasedWarmer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/FieldBasedWarmer.java @@ -34,26 +34,28 @@ public class FieldBasedWarmer extends IndexReaderWarmer { private static final Log LOG = LogFactory.getLog(FieldBasedWarmer.class); - private ShardContext shardContext; - private AtomicBoolean isClosed; + private final ShardContext _shardContext; + private final AtomicBoolean _isClosed; + private final AtomicBoolean _stop; - public FieldBasedWarmer(ShardContext shardContext, AtomicBoolean isClosed) { - this.isClosed = isClosed; - this.shardContext = shardContext; + public FieldBasedWarmer(ShardContext shardContext, AtomicBoolean isClosed, AtomicBoolean stop) { + _isClosed = isClosed; + _stop = stop; + _shardContext = shardContext; } @Override public void warm(AtomicReader reader) throws IOException { - List<String> preCacheCols = shardContext.getTableContext().getDescriptor().getPreCacheCols(); + List<String> preCacheCols = _shardContext.getTableContext().getDescriptor().getPreCacheCols(); int maxSampleSize = 1000; - IndexWarmup indexWarmup = new IndexWarmup(isClosed, maxSampleSize); - String context = shardContext.getTableContext().getTable() + "/" + shardContext.getShard(); + IndexWarmup indexWarmup = new IndexWarmup(_isClosed, _stop, maxSampleSize); + String context = _shardContext.getTableContext().getTable() + "/" + _shardContext.getShard(); Map<String, List<IndexTracerResult>> sampleIndex = indexWarmup.sampleIndex(reader, context); if (preCacheCols != null) { - warm(reader, preCacheCols, indexWarmup, sampleIndex, context, isClosed); + warm(reader, preCacheCols, indexWarmup, sampleIndex, context, _isClosed); } else { Fields fields = reader.fields(); - warm(reader, fields, indexWarmup, sampleIndex, context, isClosed); + warm(reader, fields, indexWarmup, sampleIndex, context, _isClosed); } } @@ -64,7 +66,7 @@ public class FieldBasedWarmer extends IndexReaderWarmer { indexWarmup.warm(reader, sampleIndex, field, context); } catch (IOException e) { LOG.error("Context [{0}] unknown error trying to warmup the [{1}] field", e, context, field); - LOG.error("Current sampleIndex [{0}]",sampleIndex); + LOG.error("Current sampleIndex [{0}]", sampleIndex); } if (isClosed.get()) { LOG.info("Context [{0}] index closed", context); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java b/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java index 73f66ab..bc26283 100644 --- a/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java +++ b/blur-store/src/main/java/org/apache/blur/lucene/warmup/IndexWarmup.java @@ -66,15 +66,17 @@ public class IndexWarmup { private final AtomicBoolean _isClosed; private final int _maxSampleSize; private final long _maxBytesPerSec; + private final AtomicBoolean _stop; - public IndexWarmup(AtomicBoolean isClosed, int maxSampleSize, long maxBytesPerSec) { + public IndexWarmup(AtomicBoolean isClosed, AtomicBoolean stop, int maxSampleSize, long maxBytesPerSec) { _isClosed = isClosed; + _stop = stop; _maxSampleSize = maxSampleSize; _maxBytesPerSec = maxBytesPerSec; } - public IndexWarmup(AtomicBoolean isClosed, int maxSampleSize) { - this(isClosed, maxSampleSize, DEFAULT_THROTTLE); + public IndexWarmup(AtomicBoolean isClosed, AtomicBoolean stop, int maxSampleSize) { + this(isClosed, stop, maxSampleSize, DEFAULT_THROTTLE); } private static ThreadLocal<Boolean> runTrace = new ThreadLocal<Boolean>() { @@ -137,7 +139,7 @@ public class IndexWarmup { if (endingPosition == Long.MAX_VALUE) { endingPosition = dir.fileLength(fileName) - 1; } - if (_isClosed.get()) { + if (_isClosed.get() || _stop.get()) { LOG.info("Context [{0}] index closed", context); return; } @@ -187,7 +189,7 @@ public class IndexWarmup { return getDirectory((AtomicReader) reader, segmentName, context); } for (IndexReaderContext ctext : reader.getContext().leaves()) { - if (_isClosed.get()) { + if (_isClosed.get() || _stop.get()) { LOG.info("Context [{0}] index closed", context); return null; } @@ -238,7 +240,7 @@ public class IndexWarmup { public Map<String, List<IndexTracerResult>> sampleIndex(IndexReader reader, String context) throws IOException { Map<String, List<IndexTracerResult>> results = new HashMap<String, List<IndexTracerResult>>(); for (IndexReaderContext ctext : reader.getContext().leaves()) { - if (_isClosed.get()) { + if (_isClosed.get() || _stop.get()) { LOG.info("Context [{0}] index closed", context); return null; } @@ -278,7 +280,7 @@ public class IndexWarmup { IndexTracerResult result = tracer.runTrace(terms); segmentTraces.add(result); } - if (_isClosed.get()) { + if (_isClosed.get() || _stop.get()) { LOG.info("Context [{0}] index closed", context); return null; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-util/src/main/java/org/apache/blur/trace/Trace.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/Trace.java b/blur-util/src/main/java/org/apache/blur/trace/Trace.java index 94ba2a6..be9140d 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/Trace.java +++ b/blur-util/src/main/java/org/apache/blur/trace/Trace.java @@ -109,8 +109,11 @@ public class Trace { return new Parameter(name.toString(), value.toString()); } - private static void setupTraceOnNewThread(TraceCollector parentCollector, String requestId) { - _tracer.set(new TraceCollector(parentCollector, requestId)); + private static void setupTraceOnNewThread(TraceCollector parentCollector, String requestId, int traceScope) { + TraceCollector traceCollector = new TraceCollector(parentCollector, requestId); + TracerImpl tracer = new TracerImpl(traceCollector, parentCollector.getNextId(), traceScope); + parentCollector.add(tracer); + _tracer.set(traceCollector); } private static void tearDownTraceOnNewThread() { @@ -130,7 +133,7 @@ public class Trace { if (collector == null) { return DO_NOTHING; } - TracerImpl tracer = new TracerImpl(desc, parameters, collector.getNextId()); + TracerImpl tracer = new TracerImpl(desc, parameters, collector.getNextId(), collector.getScope()); collector.add(tracer); return tracer; } @@ -150,11 +153,13 @@ public class Trace { } final long requestId = _random.get().nextLong(); Tracer trace = Trace.trace("new runnable", Trace.param(REQUEST_ID, requestId)); + TracerImpl impl = (TracerImpl) trace; + final int traceScope = impl._traceScope; try { return new Runnable() { @Override public void run() { - setupTraceOnNewThread(tc, Long.toString(requestId)); + setupTraceOnNewThread(tc, Long.toString(requestId), traceScope); Tracer t = Trace.trace("executing runnable", Trace.param(REQUEST_ID, requestId)); try { runnable.run(); @@ -176,11 +181,13 @@ public class Trace { } final long requestId = _random.get().nextLong(); Tracer trace = Trace.trace("new callable", Trace.param(REQUEST_ID, requestId)); + TracerImpl impl = (TracerImpl) trace; + final int traceScope = impl._traceScope; try { return new Callable<V>() { @Override public V call() throws Exception { - setupTraceOnNewThread(tc, Long.toString(requestId)); + setupTraceOnNewThread(tc, Long.toString(requestId), traceScope); Tracer t = Trace.trace("executing callable", Trace.param(REQUEST_ID, requestId)); try { return callable.call(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java b/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java index 68bc8b9..dbc87d3 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java +++ b/blur-util/src/main/java/org/apache/blur/trace/TraceCollector.java @@ -19,6 +19,7 @@ package org.apache.blur.trace; import java.lang.management.ManagementFactory; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import org.apache.blur.trace.Trace.TraceId; @@ -26,8 +27,9 @@ import org.apache.blur.trace.Trace.TraceId; public class TraceCollector { protected final TraceId _id; - protected final List<TracerImpl> _traces; - protected final AtomicLong _traceCounter; + protected final List<TracerImpl> _traces = new CopyOnWriteArrayList<TracerImpl>(); + protected final AtomicLong _traceCounter = new AtomicLong(); + protected final AtomicInteger _scope = new AtomicInteger(); protected final long _now = System.nanoTime(); protected final String _pid; protected final String _threadName; @@ -36,8 +38,6 @@ public class TraceCollector { public TraceCollector(String nodeName, TraceId id) { _nodeName = nodeName; _id = id; - _traces = new CopyOnWriteArrayList<TracerImpl>(); - _traceCounter = new AtomicLong(); _pid = ManagementFactory.getRuntimeMXBean().getName(); _threadName = Thread.currentThread().getName(); } @@ -45,8 +45,6 @@ public class TraceCollector { public TraceCollector(TraceCollector parentCollector, String requestId) { _nodeName = parentCollector._nodeName; _id = new TraceId(parentCollector._id.getRootId(), requestId); - _traces = parentCollector._traces; - _traceCounter = parentCollector._traceCounter; _pid = parentCollector._pid; _threadName = parentCollector._threadName; } @@ -81,4 +79,8 @@ public class TraceCollector { public long getNextId() { return _traceCounter.incrementAndGet(); } + + public AtomicInteger getScope() { + return _scope; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java b/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java index 3b855d1..83d47da 100644 --- a/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java +++ b/blur-util/src/main/java/org/apache/blur/trace/TracerImpl.java @@ -16,6 +16,8 @@ */ package org.apache.blur.trace; +import java.util.concurrent.atomic.AtomicInteger; + import org.apache.blur.trace.Trace.Parameter; public class TracerImpl implements Tracer { @@ -25,18 +27,37 @@ public class TracerImpl implements Tracer { protected long _ended; protected final String _threadName; protected final long _id; - protected Parameter[] _parameters; + protected final Parameter[] _parameters; + protected final TraceCollector _traceCollector; + protected final AtomicInteger _scope; + protected final int _traceScope; - public TracerImpl(String name, Parameter[] parameters, long id) { + public TracerImpl(String name, Parameter[] parameters, long id, AtomicInteger scope) { _name = name; _start = System.nanoTime(); _threadName = Thread.currentThread().getName(); _id = id; _parameters = parameters; + _traceCollector = null; + _scope = scope; + _traceScope = scope.incrementAndGet(); + } + + public TracerImpl(TraceCollector traceCollector, long id, int traceScope) { + _name = "new thread collector"; + _start = System.nanoTime(); + _ended = _start; + _threadName = Thread.currentThread().getName(); + _id = id; + _parameters = new Parameter[] { new Parameter("requestId", Long.toString(id)) }; + _traceCollector = traceCollector; + _scope = traceCollector.getScope(); + _traceScope = traceScope; } @Override public void done() { + _scope.decrementAndGet(); _ended = System.nanoTime(); } @@ -63,14 +84,25 @@ public class TracerImpl implements Tracer { } public String toJson() { - if (_parameters == null) { - return "{\"id\":" + _id + ", \"name\":\"" + _name + "\", \"thread\":\"" + _threadName + "\", \"took\":" - + (_ended - _start) + ", \"started\":" + _start + ", \"ended\":" + _ended + "}"; - } else { - return "{\"id\":" + _id + ", \"name\":\"" + _name + "\", \"thread\":\"" + _threadName + "\", \"took\":" - + (_ended - _start) + ", \"started\":" + _start + ", \"ended\":" + _ended + ", \"parameters\":[" - + getParametersJson() + "]}"; + StringBuilder builder = new StringBuilder(); + builder.append("{"); + builder.append("\"id\":").append(_id).append(","); + builder.append("\"scope\":").append(_traceScope).append(","); + builder.append("\"name\":\"").append(_name).append("\","); + builder.append("\"thread\":\"").append(_threadName).append("\","); + builder.append("\"took\":").append((_ended - _start)).append(","); + builder.append("\"started\":").append(_start).append(","); + builder.append("\"ended\":").append(_ended); + if (_parameters != null) { + builder.append(","); + builder.append("\"parameters\":[").append(getParametersJson()).append("]"); } + if (_traceCollector != null) { + builder.append(","); + builder.append("\"collector\":\n").append(_traceCollector.toJson()).append(""); + } + builder.append("}"); + return builder.toString(); } private String getParametersJson() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/53c40727/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java b/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java index 54a0a7f..48e1b46 100644 --- a/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java +++ b/blur-util/src/main/java/org/apache/blur/zookeeper/ZooKeeperClient.java @@ -78,7 +78,8 @@ public class ZooKeeperClient extends ZooKeeper { sessionTimeout = internalSessionTimeout; } while (true) { - Tracer trace = Trace.trace("remote call - zookeeper", Trace.param("method", executor._name)); + Tracer trace = Trace.trace("remote call - zookeeper", Trace.param("method", executor._name), + Trace.param("toString", executor.toString())); try { return executor.execute(); } catch (KeeperException e) { @@ -216,6 +217,11 @@ public class ZooKeeperClient extends ZooKeeper { LOG.debug("ZK Call - getChildren [{0}] [{1}]", path, watcher); return ZooKeeperClient.super.getChildren(path, watcher); } + + @Override + public String toString() { + return "path=" + path + " watcher=" + watcher; + } }); } @@ -227,6 +233,11 @@ public class ZooKeeperClient extends ZooKeeper { LOG.debug("ZK Call - getChildren [{0}] [{1}]", path, watch); return ZooKeeperClient.super.getChildren(path, watch); } + + @Override + public String toString() { + return "path=" + path + " watch=" + watch; + } }); }
