ISerilizableRecords

src/META-INF/explorviz.live_trace_processing.default.properties

+explorviz.live_trace_processing.worker_enabled=false
+
 explorviz.live_trace_processing.writer_target_ip=127.0.0.1
 explorviz.live_trace_processing.writer_target_port=10133
 

src/explorviz/live_trace_processing/connector/TCPConnector.java

@@ -7,10 +7,11 @@ import java.net.URL;
 import java.nio.ByteBuffer;
 import java.nio.channels.SocketChannel;
 
+import explorviz.live_trace_processing.Constants;
 import explorviz.live_trace_processing.configuration.Configuration;
 import explorviz.live_trace_processing.filter.AbstractSink;
 import explorviz.live_trace_processing.record.IRecord;
-import explorviz.live_trace_processing.record.misc.HostApplicationMetaDataRecord;
+import explorviz.live_trace_processing.record.ISerilizableRecord;
 import explorviz.live_trace_processing.writer.IWriter;
 
 public class TCPConnector extends AbstractSink implements IWriter {
@@ -18,12 +19,12 @@ public class TCPConnector extends AbstractSink implements IWriter {
 
 	private SocketChannel socketChannel;
 
-	private final Configuration configuration;
+	private final ByteBuffer buffer = ByteBuffer
+			.allocateDirect(Constants.MONITORING_MESSAGE_BUFFER_SIZE);
 
-	private ByteBuffer byteBuffer;
+	private volatile boolean shouldDisconnect = false;
 
 	public TCPConnector(final String hostname, final int port, final Configuration configuration) {
-		this.configuration = configuration;
 		try {
 			setProviderURL(new URL("http://" + hostname + ":" + port));
 		} catch (final MalformedURLException e) {
@@ -31,11 +32,6 @@ public class TCPConnector extends AbstractSink implements IWriter {
 		}
 	}
 
-	protected TCPConnector(final Configuration configuration) {
-		this.configuration = configuration;
-		providerURL = null;
-	}
-
 	@Override
 	public URL getProviderURL() {
 		return providerURL;
@@ -46,59 +42,73 @@ public class TCPConnector extends AbstractSink implements IWriter {
 		this.providerURL = providerURL;
 	}
 
-	public void init() {
-		try {
-			connect();
-		} catch (final IOException e) {
-			e.printStackTrace();
-		}
-	}
-
 	@Override
 	public void connect() throws IOException {
+		while (shouldDisconnect) {
+			try {
+				Thread.sleep(1);
+			} catch (final InterruptedException e) {
+			}
+		}
+
 		socketChannel = SocketChannel.open(new InetSocketAddress(getProviderURL().getHost(),
 				getProviderURL().getPort()));
-
-		if (socketChannel.isConnected()) {
-			// socketChannel.write(StringRegistry.getAllStringRegistryRecords());
-		}
+		// StringRegistry.sendOutAllStringRegistryRecords(); TODO
 	}
 
 	@Override
-	protected void processRecord(final IRecord record,
-			final HostApplicationMetaDataRecord hostApplicationMetaData) {
-		if (record.getRecordSizeInBytes() > byteBuffer.remaining()) {
-			send(byteBuffer);
+	protected void processRecord(final IRecord record) {
+		if (record instanceof ISerilizableRecord) {
+			final ISerilizableRecord serilizableRecord = (ISerilizableRecord) record;
+			if (buffer.remaining() < serilizableRecord.getRecordSizeInBytes()) {
+				buffer.flip();
+				send(buffer);
+				buffer.clear();
+			}
+			serilizableRecord.putIntoByteBuffer(buffer, null); // TODO
 		}
-		record.putIntoByteBuffer(byteBuffer);
 	}
 
 	private void send(final ByteBuffer buffer) {
-		if (socketChannel.isConnected()) {
+		while ((socketChannel == null) || (!socketChannel.isConnected())) {
 			try {
-				while (buffer.hasRemaining()) {
-					socketChannel.write(buffer);
-				}
-			} catch (final IOException e) {
-				System.out.println("WARNING: Connection was closed");
-				// TODO reconnect
-				disconnect();
+				Thread.sleep(1);
+			} catch (final InterruptedException e) {
 			}
 		}
-	}
 
-	@Override
-	public final void disconnect() {
-		if (socketChannel.isConnected()) {
+		try {
+			while (buffer.hasRemaining()) {
+				socketChannel.write(buffer);
+			}
+			doDisconnectIfNessecary();
+		} catch (final IOException e) {
+			System.out.println("WARNING: Connection was closed - possible data loss");
+			// TODO reconnect in non-load-balancing mode
 			try {
 				socketChannel.close();
-			} catch (final IOException e) {
-				e.printStackTrace();
+			} catch (final IOException e1) {
 			}
 		}
 	}
 
-	public void cleanup() {
-		disconnect();
+	private void doDisconnectIfNessecary() {
+		if (shouldDisconnect) {
+			if ((socketChannel != null) && socketChannel.isConnected()) {
+				try {
+					socketChannel.close();
+				} catch (final IOException e) {
+					e.printStackTrace();
+				}
+			}
+			shouldDisconnect = false;
+		}
+	}
+
+	@Override
+	public final void disconnect() {
+		if (socketChannel != null) {
+			shouldDisconnect = true;
+		}
 	}
 }

src/explorviz/live_trace_processing/filter/reconstruction/TraceReconstructionFilter.java

@@ -35,14 +35,14 @@ public final class TraceReconstructionFilter extends AbstractFilter {
 	}
 
 	@Override
-	public void processRecord(final IRecord record,
-			final HostApplicationMetaDataRecord hostApplicationMetaData) {
+	public void processRecord(final IRecord record) {
 		if (record instanceof AbstractOperationEventRecord) {
 			final AbstractOperationEventRecord abstractOperationEvent = ((AbstractOperationEventRecord) record);
 
 			final long traceId = abstractOperationEvent.getTraceId();
 			final TraceReconstructionBuffer traceBuffer = getBufferForTraceId(
-					abstractOperationEvent.getTraceId(), hostApplicationMetaData);
+					abstractOperationEvent.getTraceId(),
+					abstractOperationEvent.getHostApplicationMetadata());
 			traceBuffer.insertEvent(abstractOperationEvent);
 
 			if (traceBuffer.isFinished()) {

src/explorviz/live_trace_processing/filter/reduction/summarization/TracePatternSummarizationFilter.java

@@ -9,7 +9,6 @@ import explorviz.live_trace_processing.filter.AbstractFilter;
 import explorviz.live_trace_processing.filter.IPipeReceiver;
 import explorviz.live_trace_processing.reader.TimeProvider;
 import explorviz.live_trace_processing.record.IRecord;
-import explorviz.live_trace_processing.record.misc.HostApplicationMetaDataRecord;
 import explorviz.live_trace_processing.record.misc.TerminateRecord;
 import explorviz.live_trace_processing.record.misc.TimedPeriodRecord;
 import explorviz.live_trace_processing.record.trace.Trace;
@@ -32,8 +31,7 @@ public class TracePatternSummarizationFilter extends AbstractFilter {
 	}
 
 	@Override
-	public void processRecord(final IRecord record,
-			final HostApplicationMetaDataRecord hostApplicationMetaData) {
+	public void processRecord(final IRecord record) {
 		if (record instanceof Trace) {
 			final Trace trace = (Trace) record;
 			if (trace.isValid()) {
@@ -54,7 +52,8 @@ public class TracePatternSummarizationFilter extends AbstractFilter {
 	private void insertIntoBuffer(final Trace trace) {
 		TracePatternSummarizationBuffer traceAggregationBuffer = trace2buffer.get(trace);
 		if (traceAggregationBuffer == null) {
-			traceAggregationBuffer = new TracePatternSummarizationBuffer(TimeProvider.getCurrentTimestamp());
+			traceAggregationBuffer = new TracePatternSummarizationBuffer(
+					TimeProvider.getCurrentTimestamp());
 			trace2buffer.put(trace, traceAggregationBuffer);
 		}
 		traceAggregationBuffer.insertTrace(trace);

src/explorviz/live_trace_processing/main/WorkerStarter.java

 package explorviz.live_trace_processing.main;
 
+import java.io.IOException;
+
+import explorviz.live_trace_processing.configuration.Configuration;
+import explorviz.live_trace_processing.configuration.ConfigurationFactory;
+import explorviz.live_trace_processing.connector.TCPConnector;
 import explorviz.live_trace_processing.reader.TCPReader;
+import explorviz.live_trace_processing.writer.load_balancer.LoadBalancer;
 
 public class WorkerStarter {
 
 	public static void main(final String[] args) {
-		final TCPReader tcpReader = new TCPReader(10133, null);
-		tcpReader.read();
+		final Configuration configuration = ConfigurationFactory.createSingletonConfiguration();
+
+		final boolean worker = configuration
+				.getBooleanProperty(ConfigurationFactory.WORKER_ENABLED);
+
+		if (worker) {
+			final TCPConnector tcpConnector = new TCPConnector(
+					configuration.getStringProperty(ConfigurationFactory.WRITER_TARGET_IP),
+					configuration.getIntProperty(ConfigurationFactory.WRITER_TARGET_PORT),
+					configuration);
+
+			configureLoadBalancerIfEnabled(configuration, tcpConnector);
+
+			new TCPReader(10133, tcpConnector).read();
+		} else { // testing purpose
+			new TCPReader(10133, null).read();
+		}
+
+	}
+
+	private static void configureLoadBalancerIfEnabled(final Configuration configuration,
+			final TCPConnector tcpConnector) {
+		final boolean loadBalancerEnabled = configuration
+				.getBooleanProperty(ConfigurationFactory.LOAD_BALANCER_ENABLED);
+
+		if (loadBalancerEnabled) {
+			new LoadBalancer(
+					configuration.getStringProperty(ConfigurationFactory.LOAD_BALANCER_IP),
+					configuration.getIntProperty(ConfigurationFactory.LOAD_BALANCER_PORT),
+					configuration.getIntProperty(ConfigurationFactory.LOAD_BALANCER_WAIT_TIME),
+					configuration
+							.getStringProperty(ConfigurationFactory.LOAD_BALANCER_SCALING_GROUP),
+					configuration, tcpConnector);
+		} else {
+			try {
+				tcpConnector.connect();
+			} catch (final IOException e) {
+				e.printStackTrace();
+			}
+		}
 	}
 }

src/explorviz/live_trace_processing/reader/TCPReader.java

@@ -74,8 +74,8 @@ public final class TCPReader {
 	public final void terminate(final boolean error) {
 		System.out.println("Shutdown of TCPReader requested.");
 		active = false;
-		for (final TCPReaderOneClient thread : threads) {
-			thread.terminate();
-		}
+		// for (final TCPReaderOneClient thread : threads) {
+		// thread.terminate();
+		// }
 	}
 }

src/explorviz/live_trace_processing/reader/TCPReaderOneClient.java

@@ -62,7 +62,9 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 
 	@Override
 	public void periodicTimeSignal(final long timestamp) {
-		// TODO flush out buffer!
+		synchronized (this) { // TODO better solution
+			flushOutputBuffer();
+		}
 
 		final long hiseq = ringBuffer.next();
 		final RecordArrayEvent valueEvent = ringBuffer.get(hiseq);
@@ -70,8 +72,6 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 		buffer[0] = new TimedPeriodRecord();
 
 		valueEvent.setValues(buffer);
-		valueEvent.setValuesLength(1);
-		valueEvent.setMetadata(hostApplicationMetadata);
 
 		ringBuffer.publish(hiseq);
 	}
@@ -195,7 +195,7 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 
 		if (operation != null) {
 			putInRingBuffer(new BeforeOperationEventRecord(timestamp, traceId, orderIndex,
-					operation));
+					operation, hostApplicationMetadata));
 		} else {
 			final byte[] message = new byte[BeforeOperationEventRecord.BYTE_LENGTH_WITH_CLAZZ_ID];
 			buffer.position(buffer.position()
@@ -217,7 +217,7 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 
 		if ((operation != null) && (cause != null)) {
 			putInRingBuffer(new AfterFailedOperationEventRecord(timestamp, traceId, orderIndex,
-					operation, cause));
+					operation, cause, hostApplicationMetadata));
 		} else {
 			final byte[] message = new byte[AfterFailedOperationEventRecord.BYTE_LENGTH_WITH_CLAZZ_ID];
 			buffer.position(buffer.position()
@@ -235,7 +235,8 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 
 		final String operation = getStringFromRegistry(operationId);
 		if (operation != null) {
-			putInRingBuffer(new AfterOperationEventRecord(timestamp, traceId, orderIndex, operation));
+			putInRingBuffer(new AfterOperationEventRecord(timestamp, traceId, orderIndex,
+					operation, hostApplicationMetadata));
 		} else {
 			final byte[] message = new byte[AfterOperationEventRecord.BYTE_LENGTH_WITH_CLAZZ_ID];
 			buffer.position(buffer.position() - AfterOperationEventRecord.BYTE_LENGTH_WITH_CLAZZ_ID);
@@ -249,7 +250,8 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 		final double cpuUtil = buffer.getDouble();
 		final long usedRAM = buffer.getLong();
 		final long absoluteRAM = buffer.getLong();
-		System.out.println(new SystemMonitoringRecord(cpuUtil, usedRAM, absoluteRAM).toString());
+		System.out.println(new SystemMonitoringRecord(cpuUtil, usedRAM, absoluteRAM,
+				hostApplicationMetadata).toString());
 		// putInRingBuffer(new SystemMonitoringRecord(cpuUtil, usedRAM,
 		// absoluteRAM));
 	}
@@ -290,9 +292,11 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 	private final void putInRingBuffer(final IRecord message) {
 		counter.inputObjects(message);
 
-		outputBuffer[outputBufferIndex++] = message;
-		if (outputBufferIndex == TCPReader.OUTPUT_MESSAGE_BUFFER_SIZE) {
-			flushOutputBuffer();
+		synchronized (this) { // TODO better solution
+			outputBuffer[outputBufferIndex++] = message;
+			if (outputBufferIndex == TCPReader.OUTPUT_MESSAGE_BUFFER_SIZE) {
+				flushOutputBuffer();
+			}
 		}
 	}
 
@@ -303,8 +307,6 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 			final IRecord[] oldValues = valueEvent.getValues();
 
 			valueEvent.setValues(outputBuffer);
-			valueEvent.setValuesLength(outputBufferIndex);
-			valueEvent.setMetadata(hostApplicationMetadata);
 			ringBuffer.publish(hiseq);
 
 			outputBuffer = oldValues;
@@ -322,8 +324,4 @@ public class TCPReaderOneClient extends Thread implements IPeriodicTimeSignalRec
 	private final String getStringFromRegistry(final int id) {
 		return stringRegistry.get(id);
 	}
-
-	public void terminate() {
-		// TODO
-	}
 }