2010-09-15 Francois Chouinard <fchouinard@gmail.com> Contribution for Bug287563

[deliverable/tracecompass.git] / org.eclipse.linuxtools.lttng / src / org / eclipse / linuxtools / lttng / control / LttngSyntheticEventProvider.java

/*******************************************************************************
 * Copyright (c) 2010 Ericsson
 * 
 * All rights reserved. This program and the accompanying materials are
 * made available under the terms of the Eclipse Public License v1.0 which
 * accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 * 
 * Contributors:
 *   Alvaro Sanchez-Leon (alvsan09@gmail.com) - Initial API and implementation
 *******************************************************************************/

package org.eclipse.linuxtools.lttng.control;

import java.util.HashMap;
import java.util.Map;

import org.eclipse.linuxtools.lttng.TraceDebug;
import org.eclipse.linuxtools.lttng.event.LttngEvent;
import org.eclipse.linuxtools.lttng.event.LttngEventType;
import org.eclipse.linuxtools.lttng.event.LttngSyntheticEvent;
import org.eclipse.linuxtools.lttng.event.LttngSyntheticEvent.SequenceInd;
import org.eclipse.linuxtools.lttng.event.LttngTimestamp;
import org.eclipse.linuxtools.lttng.model.LTTngTreeNode;
import org.eclipse.linuxtools.lttng.request.LttngBaseEventRequest;
import org.eclipse.linuxtools.lttng.state.evProcessor.ITransEventProcessor;
import org.eclipse.linuxtools.lttng.state.evProcessor.state.StateEventToHandlerFactory;
import org.eclipse.linuxtools.lttng.state.model.LttngTraceState;
import org.eclipse.linuxtools.lttng.state.trace.IStateTraceManager;
import org.eclipse.linuxtools.tmf.component.TmfEventProvider;
import org.eclipse.linuxtools.tmf.event.TmfEventSource;
import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
import org.eclipse.linuxtools.tmf.experiment.TmfExperiment;
import org.eclipse.linuxtools.tmf.request.ITmfDataRequest;
import org.eclipse.linuxtools.tmf.request.ITmfEventRequest;
import org.eclipse.linuxtools.tmf.request.TmfEventRequest;
import org.eclipse.linuxtools.tmf.signal.TmfEndSynchSignal;
import org.eclipse.linuxtools.tmf.signal.TmfStartSynchSignal;
import org.eclipse.linuxtools.tmf.trace.ITmfContext;
import org.eclipse.linuxtools.tmf.trace.ITmfTrace;
import org.eclipse.linuxtools.tmf.trace.TmfContext;
import org.eclipse.linuxtools.tmf.trace.TmfTrace;

/**
 * @author alvaro
 * 
 */
public class LttngSyntheticEventProvider extends TmfEventProvider<LttngSyntheticEvent> {

	// ========================================================================
	// Data
	// ========================================================================
	public static final int BLOCK_SIZE = 1;
	public static final int NB_EVENTS  = 1;
	public static final int QUEUE_SIZE = 1; // lttng specific, one event at a
											// time

	// TmfDataProvider<LttngEvent> fExtProvider = null;
	private ITmfDataRequest<LttngSyntheticEvent> fmainRequest = null;
	private final Map<IStateTraceManager, LttngBaseEventRequest> fEventProviderRequests = new HashMap<IStateTraceManager, LttngBaseEventRequest>();
	private final LttngSyntheticEvent fStatusEvent;
	private final LttngSyntheticEvent fStatusEventAck;
	private int fMainReqEventCount = 0;
	volatile boolean startIndSent = false;
	private LTTngTreeNode fExperiment = null;
	private ITransEventProcessor fstateUpdateProcessor = StateEventToHandlerFactory.getInstance();
	private boolean waitForRequest = false;

	// ========================================================================
	// Constructor
	// ========================================================================
	/**
	 * Accessibility to package - use factory instead of this constructor
	 * 
	 * @param type
	 */
	LttngSyntheticEventProvider(Class<LttngSyntheticEvent> type) {
		super("LttngSyntheticEventProvider", type, QUEUE_SIZE);

		// prepare empty instance status indicators and allow them to travel via
		// the framework
		TmfEventSource source = new TmfEventSource(this);
		LttngEventType dtype = new LttngEventType();
		LttngTimestamp statusTimeStamp = new LttngTimestamp(TmfTimestamp.Zero);

		fStatusEvent = new LttngSyntheticEvent(null, statusTimeStamp, source, dtype, null, null, null);
		fStatusEvent.setSequenceInd(SequenceInd.STARTREQ);

		fStatusEventAck = new LttngSyntheticEvent(null, statusTimeStamp, source, dtype, null, null, null);
		fStatusEventAck.setSequenceInd(SequenceInd.ACK);
	}

	// ========================================================================
	// Methods
	// ========================================================================

	@SuppressWarnings("unchecked")
	@Override
	public ITmfContext armRequest(final ITmfDataRequest<LttngSyntheticEvent> request) {
		// validate
		// make sure we have the right type of request
		if (!(request instanceof ITmfEventRequest<?>)) {
			request.cancel();
			TraceDebug.debug("Request is not an instance of ITmfEventRequest");
			return null;
		}

		if (fExperiment == null) {
			TraceDebug.debug("Experiment is null");
			request.cancel();
			return null;
		}

		// get ready to start processing
		reset(fExperiment);

		// At least one base provider shall be available
		if (fEventProviderRequests.size() < 1) {
			request.cancel();
			TraceDebug.debug("No Base event providers available");
			return null;
		}

		fmainRequest = request;
		// define event data handling
		ITmfEventRequest<LttngSyntheticEvent> eventRequest = (ITmfEventRequest<LttngSyntheticEvent>) fmainRequest;
		TmfTimeRange reqWindow = eventRequest.getRange();

		TraceDebug.debug("Main Synthethic event request started on thread:  " + Thread.currentThread().getName());

		// loop for every traceManager in current experiment
		boolean subRequestQueued = false;
		TmfExperiment<LttngEvent> experiment = (TmfExperiment<LttngEvent>) fExperiment.getValue();
		experiment.startSynch(new TmfStartSynchSignal(0));
		for (IStateTraceManager traceManager : fEventProviderRequests.keySet()) {

			// restore trace state system to nearest check point
			TmfTimestamp checkPoint = traceManager.restoreCheckPointByTimestamp(reqWindow.getStartTime());

			// adjust start time bound to check point

			// validate so checkpoint restore is within requested bounds
			TmfTimeRange traceRange = traceManager.getTrace().getTimeRange();
			if ((checkPoint != null) && !(
					checkPoint.getValue() >= traceRange.getStartTime().getValue() &&
					checkPoint.getValue() <= traceRange.getEndTime().getValue() && 
					checkPoint.getValue() < reqWindow.getEndTime().getValue())
					) {
				// checkpoint is out of trace bounds
				continue;
			}
			TmfTimeRange adjustedRange = reqWindow;
			if (checkPoint != null) {
				adjustedRange = new TmfTimeRange(checkPoint, reqWindow.getEndTime());
			}

			LttngTraceState traceModel = traceManager.getStateModel();
			// String key = (traceManager.getTrace().getPath() +
			// traceManager.getTrace().getName()).hashCode();
			ITmfTrace trace = traceManager.getTrace();
			// create sub-request for one trace within experiment
			final LttngBaseEventRequest subRequest = new LttngBaseEventRequest(adjustedRange, reqWindow.getStartTime(),
					0, TmfEventRequest.ALL_DATA, BLOCK_SIZE, traceModel, ITmfDataRequest.ExecutionType.FOREGROUND, trace) {

				private LttngSyntheticEvent syntheticEvent = null;
				private LttngSyntheticEvent syntheticAckIndicator = null;
				long subEventCount = 0L;

				private final long fDispatchTime = getDispatchTime().getValue();
				private final LttngTraceState fTraceModel = getTraceModel();
				private LttngSyntheticEvent[] fresult = new LttngSyntheticEvent[1];

				/*
				 * (non-Javadoc)
				 * 
				 * @see org.eclipse.linuxtools.lttng.control.LttngEventRequest#handleData ()
				 */
				@Override
				public void handleData() {
					LttngEvent[] events = getData();

					// Tracer.trace("Sep: " + events[0].getTimestamp());

					if (events.length > 0) {
						handleIncomingData(events[0]);
					} else {
						TraceDebug.debug("handle data received with no data");
						// handleProviderDone(getTraceModel());
						// done();
					}
				}

				/*
				 * (non-Javadoc)
				 * 
				 * @see org.eclipse.linuxtools.tmf.request.TmfDataRequest#done()
				 */
				@Override
				public void done() {
					// mark this sub-request as completed
					super.done();
					handleProviderDone();
					// super.done();
				}

				/**
				 * Trigger the Analysis and sequential control of the events.
				 * 
				 * @param e
				 */
				private void handleIncomingData(LttngEvent e) {
					long eventTime = e.getTimestamp().getValue();

					TmfTrace<LttngEvent> inTrace = e.getParentTrace();
					if (!(inTrace == getTrace())) {
						// System.out.println("Event from a different trace discarded");
						return;
					}

					// queue the new event data and an ACK
					updateSynEvent(e);

					// If time at or above requested time, update application
					if (eventTime >= fDispatchTime) {
						// Load synthetic event as the result data
						fresult[0] = syntheticEvent;

						// BEFORE UPDATE : update to the before state-update handlers
						syntheticEvent.setSequenceInd(SequenceInd.BEFORE);
						// queueResult(syntheticEvent);
						// queueResult(syntheticAckIndicator);
						fmainRequest.setData(fresult);
						fmainRequest.handleData();
						// result[0] = syntheticAckIndicator;
						// fmainRequest.setData(result);
						// fmainRequest.handleData();

						// STATE UPDATE: local handlers
						syntheticEvent.setSequenceInd(SequenceInd.UPDATE);
						fstateUpdateProcessor.process(syntheticEvent, fTraceModel);

						// AFTER UPDATE: update to the after state-update handlers
						syntheticEvent.setSequenceInd(SequenceInd.AFTER);
						// queueResult(syntheticEvent);
						// queueResult(syntheticAckIndicator);
						fmainRequest.setData(fresult);
						fmainRequest.handleData();
						// result[0] = syntheticAckIndicator;
						// fmainRequest.setData(result);
						// fmainRequest.handleData();

						// increment once per dispatch
						incrementSynEvenCount();
						subEventCount++;
					} else {
						// event time is between checkpoint adjusted time and
						// requested time i.e. application does not expect the
						// event, however the state system needs to be re-built
						// to the dispatch point
						syntheticEvent.setSequenceInd(SequenceInd.UPDATE);
						fstateUpdateProcessor.process(syntheticEvent, fTraceModel);
					}
				}

				/**
				 * Create a synthetic event from the received new reference, if the reference is the same there is no
				 * need for a new instance
				 * 
				 * if this is the first event for this request, call start handler
				 * 
				 * @param e
				 * @return
				 */
				private LttngSyntheticEvent updateSynEvent(LttngEvent e) {
					if (syntheticEvent == null || syntheticEvent.getBaseEvent() != e) {
						syntheticEvent = new LttngSyntheticEvent(e);
						syntheticAckIndicator = new LttngSyntheticEvent(e);
						syntheticAckIndicator.setSequenceInd(SequenceInd.ACK);
					}

					// Trace model needed by application handlers
					syntheticEvent.setTraceModel(fTraceModel);

					// send the start request indication once per request thread
					if (!startIndSent) {
						TraceDebug.debug("Thread started: " + Thread.currentThread().getName());
						handleProviderStarted(getTraceModel());
						startIndSent = true;
					}

					return syntheticEvent;
				}
			};

			// preserve the associated sub request to control it e.g.
			// cancellation
			fEventProviderRequests.put(traceManager, subRequest);

			// start request
			TmfExperiment<LttngEvent> provider = (TmfExperiment<LttngEvent>) fExperiment.getValue();
			provider.sendRequest(subRequest);

			// provider.sendRequest(subRequest, ExecutionType.LONG);
			subRequestQueued = true;
		}

		experiment.endSynch(new TmfEndSynchSignal(0));

		// Return a dummy context, not used for relay provider
		return (subRequestQueued) ? new TmfContext() : null;
	}

	/**
	 * Notify listeners to prepare to receive data e.g. clean previous data etc.
	 */
	public void handleProviderStarted(LttngTraceState traceModel) {
		LttngSyntheticEvent startIndEvent = new LttngSyntheticEvent(fStatusEvent);
		startIndEvent.setSequenceInd(SequenceInd.STARTREQ);

		// Notify application
		LttngSyntheticEvent[] result = new LttngSyntheticEvent[1];
		result[0] = startIndEvent;
		fmainRequest.setData(result);
		fmainRequest.handleData();
		// result[0] = fStatusEventAck;
		// fmainRequest.setData(result);
		// fmainRequest.handleData();

		// try {
		// queueResult(startIndEvent);
		// queueResult(fStatusEventAck);
		// } catch (InterruptedException e) {
		// // TODO: cancel this request
		// e.printStackTrace();
		// }

		// Notify state event processor
		fstateUpdateProcessor.process(startIndEvent, null);
	}

	/**
	 * Notify listeners, no more events for the current request will be distributed e.g. update view.
	 */
	public synchronized void handleProviderDone() {
		// TODO: The use of a thread per main request and thread per sub-request
		// requires
		// to make sure the proper main request is marked completed. So a
		// relationship of sub-requests to parent needs to be established to
		// handle completion and cancellations properly

		// Close the main request when all sub-requests are marked completed
		for (LttngBaseEventRequest subRequest : fEventProviderRequests.values()) {
			if (subRequest != null) {
				if (!subRequest.isCompleted()) {
					// Not ready to complete main request
					return;
				}
			}
		}

		// Notify application. One notification per trace so the last state of each trace can be
		// drawn
		LttngTraceState traceModel;
		for (IStateTraceManager traceMgr : fEventProviderRequests.keySet()) {
			traceModel = traceMgr.getStateModel();

			LttngSyntheticEvent finishEvent = new LttngSyntheticEvent(fStatusEvent);
			// finishEvent.setTraceModel(traceModel);
			// finishEvent.setParentTrace((TmfTrace<LttngEvent>) traceModel.getContext().getTraceIdRef());
			finishEvent.setSequenceInd(SequenceInd.ENDREQ);
			finishEvent.setTraceModel(traceModel);

			LttngSyntheticEvent[] result = new LttngSyntheticEvent[1];
			// End Request indication
			result[0] = finishEvent;
			fmainRequest.setData(result);
			fmainRequest.handleData();

			// // End Request processed indication
			// result[0] = fStatusEventAck;
			// fmainRequest.setData(result);
			// fmainRequest.handleData();
		}

		// All sub-requests are marked completed so the main request can be completed as well
		fmainRequest.done();

		// try {
		// queueResult(finishEvent);
		// queueResult(fStatusEventAck);
		// // End the loop in the main request
		// queueResult(LttngSyntheticEvent.NullEvent);
		// } catch (InterruptedException e) {
		// // System.out.println(getName() +
		// // ":handleProviderDone() failed to queue request");
		// // TODO: Cancel the request
		// // e.printStackTrace();
		// }
	}

	/**
	 * Increment the global event counter i.e. events from any sub requests
	 */
	private synchronized void incrementSynEvenCount() {
		fMainReqEventCount++;
	}

	/**
	 * @return
	 */
	public synchronized int getSynEvenCount() {
		return fMainReqEventCount;
	}

	/**
	 * Reset provider to a state ready to begin thread execution
	 * 
	 * @param experimentNode
	 */
	public synchronized void reset(LTTngTreeNode experimentNode) {

		fmainRequest = null;

		// Make sure previous request are terminated
		for (LttngBaseEventRequest tmpRequest : fEventProviderRequests.values()) {
			if (tmpRequest != null && !tmpRequest.isCompleted()) {
				tmpRequest.cancel();
			}
		}

		fEventProviderRequests.clear();
		fMainReqEventCount = 0;
		startIndSent = false;

		// set of base event providers
		if (fExperiment != null) {
			LTTngTreeNode[] traces = fExperiment.getChildren();
			for (LTTngTreeNode trace : traces) {
				IStateTraceManager traceBaseEventProvider = (IStateTraceManager) trace;
				fEventProviderRequests.put(traceBaseEventProvider, null);
			}
		}

		if (fExperiment != experimentNode) {
			updateExperimentNode(experimentNode);
		}
	}

	/**
	 * Point to a new experiment reference
	 * 
	 * @param experiment
	 */
	private synchronized void updateExperimentNode(LTTngTreeNode experiment) {
		if (experiment != null && experiment.getValue() instanceof TmfExperiment<?>) {
			fExperiment = experiment;
		} else {
			TraceDebug
					.debug("Experiment received is not instance of TmfExperiment: " + experiment.getClass().getName());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.linuxtools.tmf.component.TmfDataProvider#sendRequest(org.
	 * eclipse.linuxtools.tmf.request.TmfDataRequest)
	 */
	@Override
	public void sendRequest(final ITmfDataRequest<LttngSyntheticEvent> request) {
		super.sendRequest(request);
		if (waitForRequest) {
			try {
				request.waitForCompletion();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * @return the waitForRequest
	 */
	public boolean isWaitForRequest() {
		return waitForRequest;
	}

	/**
	 * @param waitForRequest
	 *            configures the provider to wait for the request completion
	 */
	public void setWaitForRequest(boolean waitForRequest) {
		this.waitForRequest = waitForRequest;
	}

	@Override
	public LttngSyntheticEvent getNext(ITmfContext context) {
		try {
			fmainRequest.waitForCompletion();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return null;
	}

	// @Override
	// public LttngSyntheticEvent getNext(ITmfContext context) {
	// return super.getNext(context);
	// }

	// @Override
	// public void queueResult(LttngSyntheticEvent data) {
	// super.queueResult(data);
	// }

}