Commit 9d50ec5f authored by Florian Fittkau's avatar Florian Fittkau

WiP

parent 6d2de765
......@@ -5,7 +5,6 @@ import java.util.List;
import java.util.Stack;
import explorviz.live_trace_processing.Constants;
import explorviz.live_trace_processing.reader.TimeProvider;
import explorviz.live_trace_processing.record.event.AbstractAfterEventRecord;
import explorviz.live_trace_processing.record.event.AbstractAfterFailedEventRecord;
import explorviz.live_trace_processing.record.event.AbstractBeforeEventRecord;
......@@ -16,18 +15,19 @@ import explorviz.live_trace_processing.record.trace.RuntimeStatisticInformation;
import explorviz.live_trace_processing.record.trace.Trace;
class TraceReconstructionBuffer {
private static final int TIMEOUT_IN_SECONDS = 4;
private final List<AbstractEventRecord> events = new ArrayList<AbstractEventRecord>(
Constants.TRACE_RECONSTRUCTION_BUFFER_INITIAL_SIZE);
private int openEvents;
private volatile boolean updatedInThisPeriod = true;
private int updatedInThisPeriodCounter = TIMEOUT_IN_SECONDS;
private long lastBufferInsert = -1;
private int maxOrderIndex = -1;
public final void insertEvent(final AbstractEventRecord event) {
updatedInThisPeriod = true;
resetTimeoutCounter();
setMaxOrderIndex(event);
if ((event instanceof AbstractBeforeEventRecord)) {
......@@ -47,20 +47,16 @@ class TraceReconstructionBuffer {
events.add(event);
}
public boolean isUpdatedInThisPeriod() {
return updatedInThisPeriod;
}
public void setUpdatedInThisPeriod(final boolean updated) {
updatedInThisPeriod = updated;
public boolean isTimedout() {
return updatedInThisPeriodCounter <= 0;
}
public final long getLastBufferInsert() {
return lastBufferInsert;
public void decreaseTimeoutCounter() {
updatedInThisPeriodCounter--;
}
public final void updateLastBufferInsert() {
lastBufferInsert = TimeProvider.getCurrentTimestamp();
public void resetTimeoutCounter() {
updatedInThisPeriodCounter = TIMEOUT_IN_SECONDS;
}
private final int setMaxOrderIndex(final AbstractEventRecord event) {
......
......@@ -18,18 +18,15 @@ import explorviz.live_trace_processing.record.trace.HostApplicationMetaDataRecor
import explorviz.live_trace_processing.record.trace.Trace;
public final class TraceReconstructionFilter extends AbstractFilter implements Runnable {
private final long maxTraceTimeout;
private final Map<AbstractEventRecord, TraceReconstructionBuffer> traceIdAndHost2trace = new TreeMap<AbstractEventRecord, TraceReconstructionBuffer>(
new TraceIdAndHostComperator());
private final PipesMerger<IRecord> traceReconstructionMerger;
public TraceReconstructionFilter(final PipesMerger<IRecord> traceReconstructionMerger,
final long maxTraceTimeout, final Queue<IRecord> receiverQueue) {
final Queue<IRecord> receiverQueue) {
super(receiverQueue, "Reconstructed traces/sec", 1000);
this.traceReconstructionMerger = traceReconstructionMerger;
this.maxTraceTimeout = maxTraceTimeout;
}
@Override
......@@ -71,8 +68,6 @@ public final class TraceReconstructionFilter extends AbstractFilter implements R
}
private void checkForTimeouts(final long timestamp) {
final long traceTimeout = timestamp - maxTraceTimeout;
System.out.println("Trace Recon Filter Size: " + traceIdAndHost2trace.size());
final Iterator<Entry<AbstractEventRecord, TraceReconstructionBuffer>> iterator = traceIdAndHost2trace
......@@ -81,15 +76,11 @@ public final class TraceReconstructionFilter extends AbstractFilter implements R
while (iterator.hasNext()) {
final Entry<AbstractEventRecord, TraceReconstructionBuffer> entry = iterator.next();
final TraceReconstructionBuffer traceBuffer = entry.getValue();
if (traceBuffer.isUpdatedInThisPeriod()) {
traceBuffer.setUpdatedInThisPeriod(false);
if (!traceBuffer.isTimedout()) {
traceBuffer.decreaseTimeoutCounter();
} else {
traceBuffer.updateLastBufferInsert();
if (traceBuffer.getLastBufferInsert() <= traceTimeout) {
deliver(traceBuffer.toTrace(false));
iterator.remove();
}
deliver(traceBuffer.toTrace(false));
iterator.remove();
}
}
}
......
......@@ -21,6 +21,8 @@ import explorviz.live_trace_processing.record.trace.Trace;
import explorviz.live_trace_processing.record.trace.TraceComperator;
public class TracesSummarizationFilter extends AbstractFilter {
private static final String INVALID_TRACE = "invalid trace...";
private final long maxCollectionDuration;
private final Map<Trace, TracesSummarizationBuffer> trace2buffer = new TreeMap<Trace, TracesSummarizationBuffer>(
......@@ -50,7 +52,7 @@ public class TracesSummarizationFilter extends AbstractFilter {
insertIntoBuffer(trace);
} else {
// trace with remote records or invalid trace cant be reduced
System.out.println("invalid trace... size:" + trace.getTraceEvents().size());
System.out.println(INVALID_TRACE);
// System.out.println("invalid trace... trace:" +
// trace.toString());
makeTraceElementsAccumulator(trace);
......
......@@ -20,12 +20,12 @@ public class FilterConfiguration {
Constants.TRACE_RECONSTRUCTION_DISRUPTOR_SIZE);
new TracesSummarizationFilter(traceReductionConnector, TimeUnit.SECONDS.toNanos(3), sink)
.start();
.start();
final PipesMerger<IRecord> traceReconstructionMerger = new PipesMerger<IRecord>(
Constants.TCP_READER_DISRUPTOR_SIZE);
new TraceReconstructionFilter(traceReconstructionMerger, TimeUnit.SECONDS.toNanos(5),
new TraceReconstructionFilter(traceReconstructionMerger,
traceReductionConnector.registerProducer()).start();
new TCPReader(configuration.getIntProperty(ConfigurationFactory.READER_LISTENING_PORT,
......
......@@ -38,7 +38,7 @@ public final class TCPReader implements IPeriodicTimeSignalReceiver {
@Override
public void periodicTimeSignal(final long timestamp) {
final TimedPeriodRecord periodRecord = new TimedPeriodRecord();
while (!periodicSignalQueue.offer(periodRecord)) {
while ((periodicSignalQueue != null) && !periodicSignalQueue.offer(periodRecord)) {
try {
Thread.sleep(1);
} catch (final InterruptedException e) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment