diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java index 66b9bfd74b8bb90932b10d4242924c263091cb00..956e13abcb60cdaf16b21531fd8d1cc52c0714e3 100644 --- a/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java +++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java @@ -18,13 +18,13 @@ public class ElementThroughputMeasuringStage<T> extends ConsumerStage<T, T> { @Override protected void execute5(final T element) { - this.numPassedElements++; - this.send(element); - Long timestampInNs = this.triggerInputPort.receive(); if (timestampInNs != null) { this.computeElementThroughput(System.nanoTime()); } + this.numPassedElements++; + + this.send(element); } @Override diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/traceReconstruction/TraceReconstructionFilter.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/traceReconstruction/TraceReconstructionFilter.java index cbd26440a555efbdaa9e1c6263a3e898397135ff..8f0315efd6113b93da28309c454b3a8df99d327e 100644 --- a/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/traceReconstruction/TraceReconstructionFilter.java +++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/traceReconstruction/TraceReconstructionFilter.java @@ -15,7 +15,6 @@ ***************************************************************************/ package teetime.variant.methodcallWithPorts.stage.kieker.traceReconstruction; -import java.util.Iterator; import java.util.concurrent.TimeUnit; import teetime.util.concurrent.hashmap.ConcurrentHashMapWithDefault; @@ -57,7 +56,7 @@ public class TraceReconstructionFilter extends ConsumerStage<IFlowRecord, TraceE private void putIfFinished(final Long traceId) { final TraceBuffer traceBuffer = this.traceId2trace.get(traceId); if (traceBuffer != null && traceBuffer.isFinished()) { // null-check to check whether the trace has already been sent and removed - boolean removed = null != this.traceId2trace.remove(traceId); + boolean removed = (null != this.traceId2trace.remove(traceId)); if (removed) { this.put(traceBuffer); } @@ -83,15 +82,8 @@ public class TraceReconstructionFilter extends ConsumerStage<IFlowRecord, TraceE @Override public void onIsPipelineHead() { - synchronized (this.traceId2trace) { - Iterator<TraceBuffer> iterator = this.traceId2trace.values().iterator(); - while (iterator.hasNext()) { - TraceBuffer traceBuffer = iterator.next(); - if (traceBuffer.isFinished()) { // FIXME remove isFinished - this.put(traceBuffer); // BETTER put outside of synchronized - iterator.remove(); - } - } + for (Long traceId : this.traceId2trace.keySet()) { + this.putIfFinished(traceId); // FIXME also put invalid traces at the end } super.onIsPipelineHead(); diff --git a/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReconstruction.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReconstruction.java index 50688b496938faba3dd798c381fb4c0b9e6ae07e..085a0edfe87b0d9389042afd42508cde8b7f47c8 100644 --- a/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReconstruction.java +++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReconstruction.java @@ -26,7 +26,7 @@ import kieker.common.record.flow.IFlowRecord; public class TcpTraceReconstruction extends Analysis { private static final int NUM_VIRTUAL_CORES = Runtime.getRuntime().availableProcessors(); - private static final int TCP_RELAY_MAX_SIZE = 100000; + private static final int TCP_RELAY_MAX_SIZE = 500000; private final List<TraceEventRecords> elementCollection = new LinkedList<TraceEventRecords>(); private final ConcurrentHashMapWithDefault<Long, TraceBuffer> traceId2trace = new ConcurrentHashMapWithDefault<Long, TraceBuffer>(new TraceBuffer()); diff --git a/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReduction.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReduction.java index fb5eb3c2da1a07f61710cae0a5958b9fd529429c..c20dd7ee8a66426012d36e7b977dd84045934040 100644 --- a/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReduction.java +++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/kiekerdays/TcpTraceReduction.java @@ -32,7 +32,7 @@ import kieker.common.record.flow.IFlowRecord; public class TcpTraceReduction extends Analysis { private static final int NUM_VIRTUAL_CORES = Runtime.getRuntime().availableProcessors(); - private static final int TCP_RELAY_MAX_SIZE = 100000; + private static final int TCP_RELAY_MAX_SIZE = 500000; private final List<TraceEventRecords> elementCollection = new LinkedList<TraceEventRecords>(); private final ConcurrentHashMapWithDefault<Long, TraceBuffer> traceId2trace = new ConcurrentHashMapWithDefault<Long, TraceBuffer>(new TraceBuffer()); diff --git a/src/test/java/teetime/variant/methodcallWithPorts/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreads.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreads.java index b268f11ca0344fab6698dcfb6dd739c75b73d38b..20a52ac91455effaed083e0e6814b7736da8d0ce 100644 --- a/src/test/java/teetime/variant/methodcallWithPorts/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreads.java +++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreads.java @@ -34,7 +34,7 @@ import kieker.common.record.flow.trace.TraceMetadata; public class TcpTraceReconstructionAnalysisWithThreads extends Analysis { private static final int NUM_VIRTUAL_CORES = Runtime.getRuntime().availableProcessors(); - private static final int TCP_RELAY_MAX_SIZE = 10000000; + private static final int TCP_RELAY_MAX_SIZE = 2000000; private final List<TraceEventRecords> elementCollection = new LinkedList<TraceEventRecords>();