Commit 7eefccd4 authored by Florian Fittkau's avatar Florian Fittkau

WiP

parent 1dc8d1bf
......@@ -9,7 +9,7 @@ explorviz.live_trace_processing.writer_load_balancing_port=9999
explorviz.live_trace_processing.writer_load_balancing_wait_time=20000
explorviz.live_trace_processing.writer_load_balancing_scaling_group=analysis-worker
explorviz.live_trace_processing.sending_buffer_size=16777216
explorviz.live_trace_processing.sending_buffer_size=2097152
######################## Monitoring ########################
......@@ -38,7 +38,7 @@ explorviz.live_trace_processing.reader_listening_port=10133
explorviz.live_trace_processing.tcp_reader_disruptor_size=16384
explorviz.live_trace_processing.trace_reconstruction_disruptor_size=4096
explorviz.live_trace_processing.trace_reconstruction_disruptor_size=256
explorviz.live_trace_processing.trace_reconstruction_buffer_initial_size=128
explorviz.live_trace_processing.trace_summarization_disruptor_size=256
explorviz.live_trace_processing.trace_summarization_disruptor_size=64
......@@ -5,6 +5,8 @@ import java.util.Queue;
import explorviz.live_trace_processing.filter.AbstractFilter;
import explorviz.live_trace_processing.filter.SinglePipeConnector;
import explorviz.live_trace_processing.record.IRecord;
import explorviz.live_trace_processing.record.event.AbstractBeforeOperationEventRecord;
import explorviz.live_trace_processing.record.event.AbstractEventRecord;
import explorviz.live_trace_processing.record.misc.TerminateRecord;
import explorviz.live_trace_processing.record.misc.TimedPeriodRecord;
import explorviz.live_trace_processing.record.trace.Trace;
......@@ -29,7 +31,14 @@ public class TraceCountingFilter extends AbstractFilter {
public void processRecord(final IRecord record) {
if (record instanceof Trace) {
final Trace trace = (Trace) record;
counter.inputObjectsCount(trace.getCalledTimes());
if (!trace.getTraceEvents().isEmpty()) {
final AbstractEventRecord abstractEventRecord = trace.getTraceEvents().get(0);
if (abstractEventRecord instanceof AbstractBeforeOperationEventRecord) {
final AbstractBeforeOperationEventRecord abstractBeforeOperationEventRecord = (AbstractBeforeOperationEventRecord) abstractEventRecord;
counter.inputObjectsCount(abstractBeforeOperationEventRecord
.getRuntimeStatisticInformation().getCount());
}
}
deliver(record);
} else if (record instanceof TimedPeriodRecord) {
deliver(record);
......
......@@ -33,7 +33,10 @@ class TraceReconstructionBuffer {
if ((event instanceof AbstractBeforeEventRecord)) {
openEvents++;
final AbstractBeforeEventRecord beforeEvent = (AbstractBeforeEventRecord) event;
beforeEvent.setRuntimeStatisticInformation(new RuntimeStatisticInformation(1, -1, -1));
if (beforeEvent.getRuntimeStatisticInformation() == null) {
beforeEvent.setRuntimeStatisticInformation(new RuntimeStatisticInformation(1, -1,
-1));
}
} else if ((event instanceof AbstractAfterFailedEventRecord)
|| (event instanceof AbstractAfterEventRecord)) {
openEvents--;
......@@ -91,14 +94,16 @@ class TraceReconstructionBuffer {
if (!stack.isEmpty()) {
final AbstractBeforeEventRecord beforeEvent = stack.pop();
final long time = event.getLoggingTimestamp()
- beforeEvent.getLoggingTimestamp();
beforeEvent.setRuntimeStatisticInformation(new RuntimeStatisticInformation(1,
time, time * time));
if (beforeEvent.getRuntimeStatisticInformation() == null) {
final long time = event.getLoggingTimestamp()
- beforeEvent.getLoggingTimestamp();
beforeEvent.setRuntimeStatisticInformation(new RuntimeStatisticInformation(
1, time, time * time));
}
}
}
}
return new Trace(new ArrayList<AbstractEventRecord>(events), valid, containsRemoteRecord,
1, events.size());
events.size());
}
}
......@@ -27,8 +27,6 @@ class TracesSummarizationBuffer {
if (accumulator == null) {
accumulator = trace;
} else {
accumulator.setCalledTimes(accumulator.getCalledTimes() + trace.getCalledTimes());
final List<AbstractEventRecord> aggregatedRecords = accumulator.getTraceEvents();
final List<AbstractEventRecord> records = trace.getTraceEvents();
......@@ -42,10 +40,10 @@ class TracesSummarizationBuffer {
if (abstractBeforeEventRecord2 instanceof AbstractBeforeOperationEventRecord) {
final AbstractBeforeOperationEventRecord abstractBeforeOperationEventRecord = (AbstractBeforeOperationEventRecord) abstractBeforeEventRecord2;
abstractBeforeEventRecord
.getRuntimeStatisticInformation()
.merge(abstractBeforeOperationEventRecord
.getRuntimeStatisticInformation(),
abstractBeforeOperationEventRecord.getObjectId());
.getRuntimeStatisticInformation()
.merge(abstractBeforeOperationEventRecord
.getRuntimeStatisticInformation(),
abstractBeforeOperationEventRecord.getObjectId());
} else {
abstractBeforeEventRecord.getRuntimeStatisticInformation().merge(
abstractBeforeEventRecord2.getRuntimeStatisticInformation(), 0);
......
package explorviz.live_trace_processing.filter.reduction;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
......@@ -80,8 +81,9 @@ public class TracesSummarizationFilter extends AbstractFilter {
}
public void makeTraceElementsAccumulator(final Trace trace) {
for (int i = 0; i < trace.getTraceEvents().size(); i++) {
final AbstractEventRecord event = trace.getTraceEvents().get(i);
final List<AbstractEventRecord> traceEvents = trace.getTraceEvents();
for (final AbstractEventRecord event : traceEvents) {
if (event instanceof AbstractBeforeEventRecord) {
final AbstractBeforeEventRecord abstractBeforeEventRecord = (AbstractBeforeEventRecord) event;
......
......@@ -62,7 +62,7 @@ class TCPReaderOneClient extends Thread {
@Override
public void run() {
final ByteBuffer buffer = ByteBuffer.allocate(16 * 1024 * 1024);
final ByteBuffer buffer = ByteBuffer.allocate(2 * 1024 * 1024);
try {
String remoteAddress = "";
if (socketChannel.isConnected()) {
......@@ -194,13 +194,12 @@ class TCPReaderOneClient extends Thread {
if (containsRemoteRecordByte == (byte) 0) {
containsRemoteRecord = false;
}
final int calledTimes = buffer.getInt();
final int eventsLength = buffer.getInt();
final List<AbstractEventRecord> events = new ArrayList<AbstractEventRecord>(
eventsLength);
currentlyOpenTrace = new Trace(events, valid, containsRemoteRecord,
calledTimes, eventsLength);
eventsLength);
if (!readInTraceRecordChunks(buffer)) {
return;
......@@ -406,7 +405,11 @@ class TCPReaderOneClient extends Thread {
if (currentlyOpenTrace.getTraceEvents().size() == currentlyOpenTrace
.getEventsLength()) {
putInQueue(currentlyOpenTrace);
final List<AbstractEventRecord> traceEvents = currentlyOpenTrace
.getTraceEvents();
for (final AbstractEventRecord event : traceEvents) {
putInQueue(event);
}
currentlyOpenTrace = null;
return !cleared;
}
......
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