RemoteDataRetriever.java

Go to the documentation of this file.

package edu.rice.cs.hpc.traceviewer.db.remote;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.HashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.eclipse.core.runtime.IProgressMonitor;
import org.eclipse.core.runtime.IStatus;
import org.eclipse.core.runtime.Status;
import org.eclipse.core.runtime.jobs.Job;
import org.eclipse.swt.widgets.Shell;
import edu.rice.cs.hpc.traceviewer.data.graph.CallPath;
import edu.rice.cs.hpc.traceviewer.data.util.Constants;
import edu.rice.cs.hpc.traceviewer.data.util.Debugger;
import edu.rice.cs.hpc.traceviewer.db.remote.DecompressionThread.DecompressionItemToDo;
import edu.rice.cs.hpc.traceviewer.painter.ImageTraceAttributes;

public class RemoteDataRetriever {
    
    // -------------------------------------
    // Constants
    // -------------------------------------
    
    //For more information on message structure, see protocol documentation at the end of this file. 
    private static final int DATA = 0x44415441;
    private static final int HERE = 0x48455245;
    
    /****
     * time out counter is based on TIME_SLEEP ms unit
     */
    private static final int TIME_OUT = 2000;
    
    private static final int TIME_SLEEP = 50;
    
    // -------------------------------------
    // Variables
    // -------------------------------------

    private final Socket socket;
    DataInputStream receiver;
    BufferedInputStream rcvBacking;
    DataOutputStream sender;
    
    final int compressionType;

    /******
     * Constructor for communicating with remote data server
     * 
     * @param _serverConnection : connection socket
     * @param _statusMgr : line manager
     * @param _shell : window shell
     * @param _compressionType : type of compression, see {@link DecompressionThread.COMPRESSION_TYPE_MASK}
     * 
     * @throws IOException
     */
    public RemoteDataRetriever(Socket _serverConnection, Shell _shell, int _compressionType) throws IOException {
        socket = _serverConnection;
        
        compressionType = _compressionType;
        
        rcvBacking      = new BufferedInputStream(socket.getInputStream());
        receiver        = new DataInputStream(rcvBacking);
        
        sender          = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
    }

    //TODO: I think these are all inclusive, but check.
    public void getData( ImageTraceAttributes attributes, 
            HashMap<Integer, CallPath> _scopeMap, 
            final ConcurrentLinkedQueue<DecompressionItemToDo> workToDo) throws IOException
    {
        //Make the call
        //Check to make sure the server is sending back data
        //Wait/Receive/Parse:
                //          Make into TimeCPID[]
                //          Make into DataByRank
                //          Make into ProcessTimeline
                //          Put into appropriate place in array
        //When all are done, return the array
        // int P0, int Pn, long t0, long tn, int vertRes, int horizRes,
        int P0 = attributes.getProcessBegin();
        int Pn = attributes.getProcessEnd();
        long t0 = attributes.getTimeBegin();
        long tn = attributes.getTimeEnd();
        int vertRes = attributes.numPixelsV;
        int horizRes = attributes.numPixelsH;
        
        Debugger.printTimestampDebug("Requesting data");
        requestData(P0, Pn, t0, tn, vertRes, horizRes);
        Debugger.printTimestampDebug("Data request finished");
        
        int responseCommand = waitAndReadInt(receiver);     
        if (responseCommand != HERE)//"HERE" in ASCII
            throw new IOException("The server did not send back data");
    
        Debugger.printTimestampDebug("Data receive begin");
        
        final int ranksExpected = Math.min(Pn-P0, vertRes);

        Job unpacker = new Job("Receiving data") {

            @Override
            public IStatus run(IProgressMonitor monitor) {
                DataInputStream dataReader;
                int ranksReceived = 0;
                dataReader = receiver;
                boolean first = true;
                
                try {
                    while (ranksReceived < ranksExpected) {

                        int rankNumber = dataReader.readInt();
                        if (first){
                            Debugger.printTimestampDebug("First real data byte received.");
                            first = false;
                        }
                        int length = dataReader.readInt();// Number of CPID's

                        long startTimeForThisTimeline = dataReader.readLong();
                        long endTimeForThisTimeline = dataReader.readLong();
                        int compressedSize = dataReader.readInt();
                        
                        // when there's network issue, the value of compressedSize can be negative
                        // this has happened when the server process was suspended and the user keeps
                        //  asking data from the client to the server
                        if (compressedSize >0) {
                            byte[] compressedTraceLine = new byte[compressedSize];

                            int numRead = 0;
                            while (numRead < compressedSize) {
                                numRead += dataReader.read(compressedTraceLine,
                                        numRead, compressedSize - numRead);

                            }

                            workToDo.add(
                                new DecompressionThread.DecompressionItemToDo(
                                            compressedTraceLine, length,
                                            startTimeForThisTimeline,
                                            endTimeForThisTimeline, rankNumber,
                                            compressionType));

                            ranksReceived++;
                            monitor.worked(1);
                        }
                    }
                } catch (IOException e) {
                    //Should we provide some UI notification to the user?
                    e.printStackTrace();
                }
                monitor.done();
                //Updating the progress doesn't work anyways and will throw
                //an exception because this is a different thread
                //monitor.endProgress();
                Debugger.printTimestampDebug("Data receive end");
                return Status.OK_STATUS;
            }
        };
        unpacker.setUser(true);
        unpacker.schedule();
        
    }

    
    private void requestData(int P0, int Pn, long t0, long tn, int vertRes,
            int horizRes) throws IOException {
        sender.writeInt(DATA);
        sender.writeInt(P0);
        sender.writeInt(Pn);
        sender.writeLong(t0);
        sender.writeLong(tn);
        sender.writeInt(vertRes);
        sender.writeInt(horizRes);
        //That's it for the message
        sender.flush();
    }


    static int waitAndReadInt(DataInputStream receiver)
            throws IOException {
        int nextCommand;
        int timeout = 0;
        // Sometime the buffer is filled with 0s for some reason. This flushes
        // them out. This is awful, but otherwise we just get 0s

        while (receiver.available() <= 4
                || ((nextCommand = receiver.readInt()) == 0)) {

            if (timeout++ > TIME_OUT) {
                throw new IOException("Timeout: no response from the server.");
            }
            try {
                Thread.sleep(TIME_SLEEP);
            } catch (InterruptedException e) {

                e.printStackTrace();
            }
        }
        if (receiver.available() < 4)// There certainly isn't a message
                                        // available, since every message is at
                                        // least 4 bytes, but the next time the
                                        // buffer has anything there will be a
                                        // message
        {
            timeout = 0;
            if (timeout++ > TIME_OUT) {
                throw new IOException("Timeout while waiting for command: no response from the server.");
            }

            receiver.read(new byte[receiver.available()]);// Flush the rest of
                                                            // the buffer
            while (receiver.available() <= 0) {

                try {
                    Thread.sleep(TIME_SLEEP);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            nextCommand = receiver.readInt();
        }
        return nextCommand;
    }
    public void closeConnection() throws IOException {
        sender.writeInt(Constants.DONE);
        sender.flush();
        sender.close();
        receiver.close();
        socket.close();
    }
    
    
    static public int getTimeSleep() {
        return TIME_SLEEP;
    }
    
    static public int getTimeOut() {
        return TIME_OUT;
    }
}