simulatorpht.java

High performance DB query

/*
 * @(#)$Id: SimulatorPHT.java,v 1.3 2005/12/30 23:22:45 burkhart Exp $
 *
 * Copyright (c) 2001-2004 Regents of the University of California.
 * All rights reserved.
 *
 * This file is distributed under the terms in the attached BERKELEY-LICENSE
 * file. If you do not find these files, copies can be found by writing to:
 * Computer Science Division, Database Group, Universite of California,
 * 617 Soda Hall #1776, Berkeley, CA 94720-1776. Attention: Berkeley License
 *
 * Copyright (c) 2003-2004 Intel Corporation. All rights reserved.
 *
 * This file is distributed under the terms in the attached INTEL-LICENSE file.
 * If you do not find these files, copies can be found by writing to:
 * Intel Research Berkeley, 2150 Shattuck Avenue, Suite 1300,
 * Berkeley, CA, 94704.  Attention:  Intel License Inquiry.
 */


package pier.indexes.prefixhashtree;

import applications.overlay.OverlaySettings;
import applications.pier.PierDataQuerySettings;
import applications.pier.PierNodeMaker;
import applications.pier.PierSettings;
import execution.EndExecution;
import execution.ProgramEvent;
import execution.SimProgram;
import java.math.BigInteger;
import java.net.InetSocketAddress;
import java.util.Properties;
import java.util.Random;
import java.util.Hashtable;
import org.apache.log4j.Logger;
import overlay.location.LocationService;
import pier.components.PierBackend;
import pier.components.PierIndexManager;
import pier.exceptions.IndexTypeException;
import pier.messages.IndexMessage;
import services.LocalNode;
import services.Output;
import services.network.Payload;
import simulator.core.Event;
import simulator.core.Node;
import simulator.core.NodeMap;
import simulator.core.Simulator;
import simulator.core.SimulatorClient;
import util.BitID;
import util.logging.LogMessage;

/**
 * Class PierSim
 *
 */
public class SimulatorPHT implements SimProgram, SimulatorClient {

    private static Logger logger = Logger.getLogger(SimulatorPHT.class);
    private Properties theParams;
    private Simulator theSimulator;
    private Random myRandom;
    private NodeMap theNodeMap;
    private InetSocketAddress[] landmarkNode;
    private int curNode;
    private int randomLandmark;
    private double buildDelay;
    private int numNodes;
    private int delayedNumNodes;
    private static final int SIGNAL_JOIN = 1;
    private static final int SIGNAL_END = 8;
    private static final int ACTIVE_NODES = 0;
    private static final int LANDMARK_NODES = 3;
    
    private static final int SIGNAL_INSERT = 2;
    private static final int SIGNAL_PRINT = 4;
    private static final int SIGNAL_SEND = 5;
    private double startTime;
    private double endTime;
    private int iid;

    /**
     * Method init
     *
     * @param defaultParams
     * @param theParams
     */
    public void init(Properties defaultParams, Properties theParams) {
        this.theParams = theParams;

        OverlaySettings.setupDefaults(defaultParams);
        PierSettings.setupDefaults(defaultParams);
        PierDataQuerySettings.setupDefaults(defaultParams);
        
        //Output.setDebugging(true);
    }

    /**
     * Method run
     *
     * @param theSimulator
     * @param theNodeMap
     * @param random
     */
    public void run(Simulator theSimulator, NodeMap theNodeMap, Random random) {
        this.theSimulator = theSimulator;
        this.theNodeMap = theNodeMap;
        this.myRandom = random;
        this.randomLandmark =
            Integer.parseInt(theParams.getProperty("simRandomLandmark"));

        int landmarkNodes = ((randomLandmark > 0)
                             ? randomLandmark
                             : 1);

        landmarkNode = new InetSocketAddress[landmarkNodes];
        this.numNodes = Integer.parseInt(theParams.getProperty("simNumNodes"));
        this.delayedNumNodes =
            Integer.parseInt(theParams.getProperty("simDelayedNumNodes"));
        this.buildDelay =
            Double.parseDouble(theParams.getProperty("buildDelay"));

        this.startTime =
            Double.parseDouble(theParams.getProperty("startTime"));
        this.endTime = Double.parseDouble(theParams.getProperty("endTime"));

        if (startTime >= 0) {
            theSimulator.schedule(ProgramEvent.allocate(this, 0, startTime,
                                                        0, SIGNAL_JOIN));
        }

        int numInserts = 4;  // default
        try {
        	numInserts = Integer.parseInt(theParams.getProperty("numInserts"));
        } catch (Exception e) {
        	// If it doesn't work, just use the default above
        }
        
        this.iid = 0;
        int i = 1;
        while (i <= numInserts) {
        	scheduleEvent(i*100, SIGNAL_INSERT, 0);
        	i++;
        }
        
        scheduleEvent((i+3)*100, SIGNAL_PRINT, 0);
        
        scheduleEvent((i+3)*100, SIGNAL_SEND, 0);
        
        if (endTime >= 0) {
            theSimulator.schedule(ProgramEvent.allocate(this, 0, endTime, -1,
                                                        SIGNAL_END));
        }
    }

    /**
     * Method loadContext
     * 
     * @param tokens
     * @param pos
     * @param noRandom
     * @return
     */
    private int loadContext(String[] tokens, int pos, boolean noRandom) {
        if ((tokens != null) && (tokens.length > pos)) {
            Node theNode = theNodeMap.getNode(
                               PierNodeMaker.parseAddressString(tokens[pos]));
            int theNodeNum = theNode.getNodeMapPosition();

            theSimulator.loadNode(theNode);

            return theNodeNum;
        }

        if (noRandom) {
            return -1;
        }

        // Chose a random active node
        int theNodeNum = theNodeMap.randomMember(ACTIVE_NODES,
                                                 myRandom.nextInt());

        theSimulator.loadNode(theNodeNum);

        return theNodeNum;
    }

    /**
     * Method scheduleEvent
     * 
     * @param delay
     * @param signal
     * @param node
     */
    private void scheduleEvent(double delay, int signal, int node) {
        double curTime = theSimulator.getCurrentTime();
        double nextTime = curTime + delay;

        theSimulator.schedule(ProgramEvent.allocate(this, curTime, nextTime,
                                                    node, signal));
    }

    /**
     * Method join
     */
    private void join() {
        theNodeMap.setMember(curNode, ACTIVE_NODES, true);
        ((LocationService) LocalNode.myGlobals[PierNodeMaker.LOCATION_SERVICE_POS]).join(
            landmarkNode);

        if (Output.debuggingEnabled) {
            for (int i = 0; i < landmarkNode.length; i++) {
                logger.debug(new LogMessage(new Object[]{"Node ",
                                                         String.valueOf(
                                                             curNode),
                                                         " joining via ",
                                                         landmarkNode[i]}));
            }
        }

        if ((landmarkNode[0] == null) || (randomLandmark > 0)) {
            int landmarkNodes = ((randomLandmark > 0)
                                 ? randomLandmark
                                 : 1);

            for (int i = 0; i < landmarkNodes; i++) {
                int nodeNum = theNodeMap.randomMember(ACTIVE_NODES,
                                                      myRandom.nextInt());
                Node nextLandmark = theNodeMap.getNode(nodeNum);

                landmarkNode[i] = new InetSocketAddress(
                    nextLandmark.getIPAddress(),
                    Integer.parseInt(
                        theParams.getProperty("locationPortNumber")));

                theNodeMap.setMember(nodeNum, LANDMARK_NODES, true);
            }
        }

        curNode++;

        if (curNode < (numNodes - delayedNumNodes)) {
            double curTime = theSimulator.getCurrentTime();
            double nextJoinTime = curTime + buildDelay;

            theSimulator.schedule(ProgramEvent.allocate(this, curTime,
                                                        nextJoinTime, curNode,
                                                        SIGNAL_JOIN));
        }
    }
    
    /**
     * Method insert
     */
    private void insert() {

    	PierBackend firstNode = (PierBackend) LocalNode.myGlobals[PierNodeMaker.PIERSERVER_POS];
    	PierIndexManager indexManager = firstNode.getIndexManager();
    	
    	Random myRand = new Random();
    	
    	try {
    		
    		int numBits = 16;    // default
    		int numPerInsert = 4;  // default
    		try {
    			numBits = Integer.parseInt(theParams.getProperty("idBits"));
    			numPerInsert = Integer.parseInt(theParams.getProperty("numPerInsert"));
    		} catch (Exception e) {
    			// Don't worry about it, just use the defaults!
    		}
    		
    		for (int i = 0; i < numPerInsert; i++) {
    			BitID key = new BitID(new BigInteger(numBits, myRand));
    			BitID value = new BitID(new BigInteger(new byte[]{(byte)i}));
    			indexManager.insert("pier.indexes.prefixhashtree.PrefixHashTree", "PHT", 
    					key, value, this.iid++, ((long)endTime)*1000);
    		}
    		
    	} catch (IndexTypeException ite) {
    		throw new PHTException("Incorrect index type passed to insert in SimulatorPHT!");
    	}
    }
    
    /**
     * Method print
     */
    private void print() {
    	PierBackend firstNode = (PierBackend) LocalNode.myGlobals[PierNodeMaker.PIERSERVER_POS];
    	PierIndexManager indexManager = firstNode.getIndexManager();
    	try {
    		indexManager.printTree("pier.indexes.prefixhashtree.PrefixHashTree", "PHT");
    	} catch (IndexTypeException ite) {
    		throw new PHTException("Incorrect index type passed to print in SimulatorPHT!");
    	}
    }
    
    /**
     * Method send
     */
    private void send() {

    	PierBackend firstNode = (PierBackend) LocalNode.myGlobals[PierNodeMaker.PIERSERVER_POS];
    	PierIndexManager indexManager = firstNode.getIndexManager();
    	
    	Random myRand = new Random();
    	
    	try {
    		
    		int numBits = 16;    // default
    		try {
    			numBits = Integer.parseInt(theParams.getProperty("idBits"));
    		} catch (Exception e) {
    			// Don't worry about it, just use the default!
    		}

			BigInteger key1bigint = new BigInteger(numBits, myRand);
			BigInteger key2bigint = new BigInteger(numBits, myRand);

			Payload[] keys = new Payload[2];
			if (key1bigint.compareTo(key2bigint) <= 0) {
				keys[0] = new BitID(key1bigint);
				keys[1] = new BitID(key2bigint);
			} else {
				keys[0] = new BitID(key2bigint);
				keys[1] = new BitID(key1bigint);
			}
			
    		indexManager.send("pier.indexes.prefixhashtree.PrefixHashTree", "PHT", keys, 
    				IndexMessage.allocate("pier.indexes.prefixhashtree.PrefixHashTree", "PHT", 100000, 
    						firstNode.getProvider().getNodeSocketAddress(), 1, new BitID()), "PHT-NS", "RID", 
							0, 100000);
    		
    	} catch (IndexTypeException ite) {
    		throw new PHTException("Incorrect index type passed to insert in SimulatorPHT!");
    	}
    }

    /**
     * Method handleEvent
     *
     * @param theEvent
     */
    public void handleEvent(Event theEvent) {
        ProgramEvent event = (ProgramEvent) theEvent;

        if (event.getSignal() == SIGNAL_JOIN) {
            join();
        }
        
        if (event.getSignal() == SIGNAL_INSERT) {
        	insert();
        }
        if (event.getSignal() == SIGNAL_PRINT) {
        	print();
        }
        if (event.getSignal() == SIGNAL_SEND) {
        	send();
        }
      
        if (event.getSignal() == SIGNAL_END) {
        	System.out.println("COMPLETE");
            throw new EndExecution("COMPLETE");
        }
    }
}