[deliverable/tracecompass.git] / org.eclipse.linuxtools.lttng / src / org / eclipse / linuxtools / lttng / trace / LTTngExperiment.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:
 *   Francois Chouinard - Initial API and implementation
 *******************************************************************************/

package org.eclipse.linuxtools.lttng.trace;

import org.eclipse.linuxtools.lttng.event.LttngEvent;
import org.eclipse.linuxtools.lttng.event.LttngTimestamp;
import org.eclipse.linuxtools.lttng.jni.JniTrace;
import org.eclipse.linuxtools.tmf.event.TmfEvent;
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.experiment.TmfExperimentContext;
import org.eclipse.linuxtools.tmf.experiment.TmfExperimentLocation;
import org.eclipse.linuxtools.tmf.request.ITmfDataRequest;
import org.eclipse.linuxtools.tmf.request.ITmfDataRequest.ExecutionType;
import org.eclipse.linuxtools.tmf.request.TmfEventRequest;
import org.eclipse.linuxtools.tmf.signal.TmfExperimentRangeUpdatedSignal;
import org.eclipse.linuxtools.tmf.signal.TmfSignalHandler;
import org.eclipse.linuxtools.tmf.signal.TmfSignalManager;
import org.eclipse.linuxtools.tmf.trace.ITmfTrace;
import org.eclipse.linuxtools.tmf.trace.TmfContext;

/**
 * <b><u>LTTngExperiment</u></b>
 * <p>
 * Temporary class to resolve a basic incompatibility between TMF and LTTng.
 * <p>
 */
public class LTTngExperiment<T extends TmfEvent> extends TmfExperiment<T> implements ITmfTrace {

    private static final int DEFAULT_INDEX_PAGE_SIZE = 50000;

    // ------------------------------------------------------------------------
    // Constructors
    // ------------------------------------------------------------------------

    /**
     * @param type
     * @param id
     * @param traces
     * @param epoch
     * @param indexPageSize
     */
    public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces, TmfTimestamp epoch, int indexPageSize) {
        this(type, id, traces, TmfTimestamp.Zero, indexPageSize, false);
    }

    public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces, TmfTimestamp epoch, int indexPageSize, boolean preIndexExperiment) {
        super(type, id, traces, epoch, indexPageSize, preIndexExperiment);
    }

    /**
     * @param type
     * @param id
     * @param traces
     */
    public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces) {
        this(type, id, traces, TmfTimestamp.Zero, DEFAULT_INDEX_PAGE_SIZE);
    }

    /**
     * @param type
     * @param id
     * @param traces
     * @param indexPageSize
     */
    public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces, int indexPageSize) {
        this(type, id, traces, TmfTimestamp.Zero, indexPageSize);
    }

    public LTTngExperiment(LTTngExperiment<T> other) {
        super(other.getName() + "(clone)", other.fType); //$NON-NLS-1$

        fEpoch = other.fEpoch;
        fIndexPageSize = other.fIndexPageSize;

        fTraces = new ITmfTrace[other.fTraces.length];
        for (int trace = 0; trace < other.fTraces.length; trace++) {
            fTraces[trace] = other.fTraces[trace].createTraceCopy();
        }

        fNbEvents = other.fNbEvents;
        fTimeRange = other.fTimeRange;
    }

    @Override
    public LTTngExperiment<T> createTraceCopy() {
        LTTngExperiment<T> experiment = new LTTngExperiment<T>(this);
        TmfSignalManager.deregister(experiment);
        return experiment;
    }

    // ------------------------------------------------------------------------
    // ITmfTrace trace positioning
    // ------------------------------------------------------------------------

    @Override
    public synchronized TmfEvent getNextEvent(TmfContext context) {

        // Validate the context
        if (!(context instanceof TmfExperimentContext)) {
            return null; // Throw an exception?
        }

        if (!context.equals(fExperimentContext)) {
//    		Tracer.trace("Ctx: Restoring context");
            fExperimentContext = seekLocation(context.getLocation());
        }

        TmfExperimentContext expContext = (TmfExperimentContext) context;

//		dumpContext(expContext, true);

        // If an event was consumed previously, get the next one from that trace
        int lastTrace = expContext.getLastTrace();
        if (lastTrace != TmfExperimentContext.NO_TRACE) {
            TmfContext traceContext = expContext.getContexts()[lastTrace];
            expContext.getEvents()[lastTrace] = expContext.getTraces()[lastTrace].getNextEvent(traceContext);
            expContext.setLastTrace(TmfExperimentContext.NO_TRACE);
        }

        // Scan the candidate events and identify the "next" trace to read from
        TmfEvent eventArray[] = expContext.getEvents();
        if (eventArray == null) {
            return null;
        }
        int trace = TmfExperimentContext.NO_TRACE;
        TmfTimestamp timestamp = TmfTimestamp.BigCrunch;
        if (eventArray.length == 1) {
            if (eventArray[0] != null) {
                timestamp = eventArray[0].getTimestamp();
                trace = 0;
            }
        } else {
            for (int i = 0; i < eventArray.length; i++) {
                TmfEvent event = eventArray[i];
                if (event != null && event.getTimestamp() != null) {
                    TmfTimestamp otherTS = event.getTimestamp();
                    if (otherTS.compareTo(timestamp, true) < 0) {
                        trace = i;
                        timestamp = otherTS;
                    }
                }
            }
        }

        // Update the experiment context and set the "next" event
        TmfEvent event = null;
        if (trace != TmfExperimentContext.NO_TRACE) {
//	        updateIndex(expContext, timestamp);

            TmfContext traceContext = expContext.getContexts()[trace];
            TmfExperimentLocation expLocation = (TmfExperimentLocation) expContext.getLocation();
            expLocation.getLocation().locations[trace] = traceContext.getLocation();

            updateIndex(expContext, timestamp);

            expLocation.getRanks()[trace] = traceContext.getRank();
            expContext.setLastTrace(trace);
            expContext.updateRank(1);
            event = expContext.getEvents()[trace];
            fExperimentContext = expContext;
        }

//		if (event != null) {
//    		Tracer.trace("Exp: " + (expContext.getRank() - 1) + ": " + event.getTimestamp().toString());
//    		dumpContext(expContext, false);
//    		Tracer.trace("Ctx: Event returned= " + event.getTimestamp().toString());
//		}

        return event;
    }

    @SuppressWarnings("unchecked")
    @Override
    protected void indexExperiment(final boolean waitForCompletion) {
        if (waitForCompletion) {
            TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
                    TmfTimeRange.Eternity);
            broadcast(signal);
            while (isIndexingBusy()) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            ;
            return;
        }
        for (ITmfTrace trace : fTraces) {
            if (trace instanceof LTTngTrace) {
                JniTrace jniTrace = ((LTTngTrace) trace).getCurrentJniTrace();
                if (jniTrace != null && !jniTrace.isLiveTraceSupported()) {
                    updateTimeRange();
                    TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
                            getTimeRange());
                    broadcast(signal);
                    return;
                }
            }
        }
        final Thread thread = new Thread("Streaming Monitor for " + getName()) { //$NON-NLS-1$
            LttngTimestamp safeTimestamp = null;
            TmfTimeRange timeRange = null;

            @Override
            public void run() {
                while (!fExecutor.isShutdown()) {
                    final TmfEventRequest<LttngEvent> request = new TmfEventRequest<LttngEvent>(LttngEvent.class, TmfTimeRange.Eternity, 0,
                            ExecutionType.FOREGROUND) {
                        @Override
                        public void handleCompleted() {
                            super.handleCompleted();
                            if (isIndexingBusy()) {
                                timeRange = null;
                                return;
                            }
                            long startTime = Long.MAX_VALUE;
                            long endTime = Long.MIN_VALUE;
                            for (ITmfTrace trace : getTraces()) {
                                if (trace instanceof LTTngTrace) {
                                    LTTngTrace lttngTrace = (LTTngTrace) trace;
                                    JniTrace jniTrace = lttngTrace.getCurrentJniTrace();
                                    jniTrace.updateTrace();
                                    startTime = Math.min(startTime, lttngTrace.getStartTime().getValue());
                                    endTime = Math.max(endTime, jniTrace.getEndTime().getTime());
                                }
                            }
                            LttngTimestamp startTimestamp = new LttngTimestamp(startTime);
                            LttngTimestamp endTimestamp = new LttngTimestamp(endTime);
                            if (safeTimestamp != null && safeTimestamp.compareTo(getTimeRange().getEndTime(), false) > 0) {
                                timeRange = new TmfTimeRange(startTimestamp, safeTimestamp);
                            } else {
                                timeRange = null;
                            }
                            safeTimestamp = endTimestamp;
                        }
                    };
                    try {
                        sendRequest((ITmfDataRequest<T>) request);
                        request.waitForCompletion();
                        if (timeRange != null && !timeRange.equals(TmfTimeRange.Null)) {
                            TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
                                    timeRange);
                            broadcast(signal);
                        }
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        thread.start();
    }

    @TmfSignalHandler
    public void experimentRangeUpdated(TmfExperimentRangeUpdatedSignal signal) {
        indexExperiment(false, (int) fNbEvents, signal.getRange());
    }

    /*
     * (non-Javadoc)
     * 
     * @see java.lang.Object#toString()
     */
    @Override
    @SuppressWarnings("nls")
    public String toString() {
        return "[LTTngExperiment (" + getName() + ")]";
    }

}
Commit	Line	Data
82e04272 FC	1	/*******************************************************************************
	2	* Copyright (c) 2010 Ericsson
	3	*
	4	* All rights reserved. This program and the accompanying materials are
	5	* made available under the terms of the Eclipse Public License v1.0 which
	6	* accompanies this distribution, and is available at
	7	* http://www.eclipse.org/legal/epl-v10.html
	8	*
	9	* Contributors:
	10	* Francois Chouinard - Initial API and implementation
	11	*******************************************************************************/
	12
	13	package org.eclipse.linuxtools.lttng.trace;
	14
a79913eb FC	15	import org.eclipse.linuxtools.lttng.event.LttngEvent;
	16	import org.eclipse.linuxtools.lttng.event.LttngTimestamp;
	17	import org.eclipse.linuxtools.lttng.jni.JniTrace;
82e04272	18	import org.eclipse.linuxtools.tmf.event.TmfEvent;
a79913eb	19	import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
82e04272 FC	20	import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
	21	import org.eclipse.linuxtools.tmf.experiment.TmfExperiment;
	22	import org.eclipse.linuxtools.tmf.experiment.TmfExperimentContext;
	23	import org.eclipse.linuxtools.tmf.experiment.TmfExperimentLocation;
a79913eb FC	24	import org.eclipse.linuxtools.tmf.request.ITmfDataRequest;
	25	import org.eclipse.linuxtools.tmf.request.ITmfDataRequest.ExecutionType;
	26	import org.eclipse.linuxtools.tmf.request.TmfEventRequest;
	27	import org.eclipse.linuxtools.tmf.signal.TmfExperimentRangeUpdatedSignal;
	28	import org.eclipse.linuxtools.tmf.signal.TmfSignalHandler;
82e04272 FC	29	import org.eclipse.linuxtools.tmf.signal.TmfSignalManager;
	30	import org.eclipse.linuxtools.tmf.trace.ITmfTrace;
	31	import org.eclipse.linuxtools.tmf.trace.TmfContext;
	32
	33	/**
	34	* <b><u>LTTngExperiment</u></b>
	35	* <p>
	36	* Temporary class to resolve a basic incompatibility between TMF and LTTng.
	37	* <p>
	38	*/
	39	public class LTTngExperiment<T extends TmfEvent> extends TmfExperiment<T> implements ITmfTrace {
	40
a79913eb	41	private static final int DEFAULT_INDEX_PAGE_SIZE = 50000;
82e04272	42
a79913eb	43	// ------------------------------------------------------------------------
82e04272 FC	44	// Constructors
	45	// ------------------------------------------------------------------------
	46
	47	/**
	48	* @param type
	49	* @param id
	50	* @param traces
	51	* @param epoch
	52	* @param indexPageSize
	53	*/
	54	public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces, TmfTimestamp epoch, int indexPageSize) {
	55	this(type, id, traces, TmfTimestamp.Zero, indexPageSize, false);
a79913eb	56	}
82e04272 FC	57
82e04272 FC	58	public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces, TmfTimestamp epoch, int indexPageSize, boolean preIndexExperiment) {
a79913eb FC	59	super(type, id, traces, epoch, indexPageSize, preIndexExperiment);
a79913eb FC	60	}
82e04272 FC	61
	62	/**
	63	* @param type
	64	* @param id
	65	* @param traces
	66	*/
	67	public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces) {
	68	this(type, id, traces, TmfTimestamp.Zero, DEFAULT_INDEX_PAGE_SIZE);
	69	}
	70
	71	/**
	72	* @param type
	73	* @param id
	74	* @param traces
	75	* @param indexPageSize
	76	*/
	77	public LTTngExperiment(Class<T> type, String id, ITmfTrace[] traces, int indexPageSize) {
	78	this(type, id, traces, TmfTimestamp.Zero, indexPageSize);
	79	}
a79913eb	80
82e04272	81	public LTTngExperiment(LTTngExperiment<T> other) {
a79913eb FC	82	super(other.getName() + "(clone)", other.fType); //$NON-NLS-1$
	83
	84	fEpoch = other.fEpoch;
	85	fIndexPageSize = other.fIndexPageSize;
	86
	87	fTraces = new ITmfTrace[other.fTraces.length];
	88	for (int trace = 0; trace < other.fTraces.length; trace++) {
	89	fTraces[trace] = other.fTraces[trace].createTraceCopy();
	90	}
	91
	92	fNbEvents = other.fNbEvents;
	93	fTimeRange = other.fTimeRange;
	94	}
	95
	96	@Override
	97	public LTTngExperiment<T> createTraceCopy() {
	98	LTTngExperiment<T> experiment = new LTTngExperiment<T>(this);
	99	TmfSignalManager.deregister(experiment);
	100	return experiment;
82e04272	101	}
a79913eb FC	102
a79913eb FC	103	// ------------------------------------------------------------------------
82e04272 FC	104	// ITmfTrace trace positioning
	105	// ------------------------------------------------------------------------
	106
a79913eb FC	107	@Override
a79913eb FC	108	public synchronized TmfEvent getNextEvent(TmfContext context) {
82e04272	109
a79913eb FC	110	// Validate the context
	111	if (!(context instanceof TmfExperimentContext)) {
	112	return null; // Throw an exception?
	113	}
82e04272	114
a79913eb	115	if (!context.equals(fExperimentContext)) {
82e04272	116	// Tracer.trace("Ctx: Restoring context");
a79913eb FC	117	fExperimentContext = seekLocation(context.getLocation());
	118	}
	119
	120	TmfExperimentContext expContext = (TmfExperimentContext) context;
82e04272 FC	121
	122	// dumpContext(expContext, true);
	123
a79913eb FC	124	// If an event was consumed previously, get the next one from that trace
	125	int lastTrace = expContext.getLastTrace();
	126	if (lastTrace != TmfExperimentContext.NO_TRACE) {
	127	TmfContext traceContext = expContext.getContexts()[lastTrace];
	128	expContext.getEvents()[lastTrace] = expContext.getTraces()[lastTrace].getNextEvent(traceContext);
	129	expContext.setLastTrace(TmfExperimentContext.NO_TRACE);
	130	}
	131
	132	// Scan the candidate events and identify the "next" trace to read from
	133	TmfEvent eventArray[] = expContext.getEvents();
	134	if (eventArray == null) {
	135	return null;
	136	}
	137	int trace = TmfExperimentContext.NO_TRACE;
	138	TmfTimestamp timestamp = TmfTimestamp.BigCrunch;
	139	if (eventArray.length == 1) {
	140	if (eventArray[0] != null) {
	141	timestamp = eventArray[0].getTimestamp();
	142	trace = 0;
	143	}
	144	} else {
	145	for (int i = 0; i < eventArray.length; i++) {
	146	TmfEvent event = eventArray[i];
	147	if (event != null && event.getTimestamp() != null) {
	148	TmfTimestamp otherTS = event.getTimestamp();
	149	if (otherTS.compareTo(timestamp, true) < 0) {
	150	trace = i;
	151	timestamp = otherTS;
	152	}
	153	}
	154	}
	155	}
	156
	157	// Update the experiment context and set the "next" event
	158	TmfEvent event = null;
	159	if (trace != TmfExperimentContext.NO_TRACE) {
82e04272 FC	160	// updateIndex(expContext, timestamp);
82e04272 FC	161
a79913eb FC	162	TmfContext traceContext = expContext.getContexts()[trace];
	163	TmfExperimentLocation expLocation = (TmfExperimentLocation) expContext.getLocation();
	164	expLocation.getLocation().locations[trace] = traceContext.getLocation();
82e04272	165
a79913eb	166	updateIndex(expContext, timestamp);
82e04272	167
a79913eb FC	168	expLocation.getRanks()[trace] = traceContext.getRank();
	169	expContext.setLastTrace(trace);
	170	expContext.updateRank(1);
	171	event = expContext.getEvents()[trace];
	172	fExperimentContext = expContext;
	173	}
82e04272 FC	174
	175	// if (event != null) {
	176	// Tracer.trace("Exp: " + (expContext.getRank() - 1) + ": " + event.getTimestamp().toString());
	177	// dumpContext(expContext, false);
	178	// Tracer.trace("Ctx: Event returned= " + event.getTimestamp().toString());
	179	// }
	180
a79913eb FC	181	return event;
	182	}
	183
	184	@SuppressWarnings("unchecked")
	185	@Override
	186	protected void indexExperiment(final boolean waitForCompletion) {
	187	if (waitForCompletion) {
	188	TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
	189	TmfTimeRange.Eternity);
	190	broadcast(signal);
	191	while (isIndexingBusy()) {
	192	try {
	193	Thread.sleep(100);
	194	} catch (InterruptedException e) {
	195	e.printStackTrace();
	196	}
	197	}
	198	;
	199	return;
	200	}
	201	for (ITmfTrace trace : fTraces) {
	202	if (trace instanceof LTTngTrace) {
	203	JniTrace jniTrace = ((LTTngTrace) trace).getCurrentJniTrace();
	204	if (jniTrace != null && !jniTrace.isLiveTraceSupported()) {
	205	updateTimeRange();
	206	TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
	207	getTimeRange());
	208	broadcast(signal);
	209	return;
	210	}
	211	}
	212	}
	213	final Thread thread = new Thread("Streaming Monitor for " + getName()) { //$NON-NLS-1$
	214	LttngTimestamp safeTimestamp = null;
	215	TmfTimeRange timeRange = null;
82e04272	216
a79913eb FC	217	@Override
	218	public void run() {
	219	while (!fExecutor.isShutdown()) {
	220	final TmfEventRequest<LttngEvent> request = new TmfEventRequest<LttngEvent>(LttngEvent.class, TmfTimeRange.Eternity, 0,
	221	ExecutionType.FOREGROUND) {
	222	@Override
	223	public void handleCompleted() {
	224	super.handleCompleted();
	225	if (isIndexingBusy()) {
	226	timeRange = null;
	227	return;
	228	}
	229	long startTime = Long.MAX_VALUE;
	230	long endTime = Long.MIN_VALUE;
	231	for (ITmfTrace trace : getTraces()) {
	232	if (trace instanceof LTTngTrace) {
	233	LTTngTrace lttngTrace = (LTTngTrace) trace;
	234	JniTrace jniTrace = lttngTrace.getCurrentJniTrace();
	235	jniTrace.updateTrace();
74237cc3	236	startTime = Math.min(startTime, lttngTrace.getStartTime().getValue());
a79913eb FC	237	endTime = Math.max(endTime, jniTrace.getEndTime().getTime());
	238	}
	239	}
	240	LttngTimestamp startTimestamp = new LttngTimestamp(startTime);
	241	LttngTimestamp endTimestamp = new LttngTimestamp(endTime);
	242	if (safeTimestamp != null && safeTimestamp.compareTo(getTimeRange().getEndTime(), false) > 0) {
	243	timeRange = new TmfTimeRange(startTimestamp, safeTimestamp);
	244	} else {
	245	timeRange = null;
	246	}
	247	safeTimestamp = endTimestamp;
	248	}
	249	};
	250	try {
	251	sendRequest((ITmfDataRequest<T>) request);
	252	request.waitForCompletion();
b626c6f7	253	if (timeRange != null && !timeRange.equals(TmfTimeRange.Null)) {
a79913eb FC	254	TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
	255	timeRange);
	256	broadcast(signal);
	257	}
74237cc3	258	Thread.sleep(2000);
a79913eb FC	259	} catch (InterruptedException e) {
	260	e.printStackTrace();
	261	}
	262	}
	263	}
	264	};
	265	thread.start();
	266	}
	267
	268	@TmfSignalHandler
	269	public void experimentRangeUpdated(TmfExperimentRangeUpdatedSignal signal) {
	270	indexExperiment(false, (int) fNbEvents, signal.getRange());
	271	}
	272
	273	/*
	274	* (non-Javadoc)
	275	*
	276	* @see java.lang.Object#toString()
	277	*/
	278	@Override
3b38ea61	279	@SuppressWarnings("nls")
a79913eb FC	280	public String toString() {
	281	return "[LTTngExperiment (" + getName() + ")]";
	282	}
82e04272 FC	283
82e04272 FC	284	}