kingnetwork.java

High performance DB query

/*
 * @(#)$Id: KingNetwork.java,v 1.7 2005/09/06 21:03:43 huebsch 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 simulator.schedulers.network.topology.king;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.StreamTokenizer;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import org.apache.log4j.Logger;
import services.Output;
import simulator.schedulers.network.NetworkTopology;
import util.logging.StructuredLogMessage;

/**
 * Class HomogenousNetwork
 * Based on Sean Rhea's King Network parsing code, bamboo/src/bamboo/sim/KingNetworkModel.java
 * Find data set at http://www.pdos.lcs.mit.edu/p2psim/kingdata
 */
public class KingNetwork implements NetworkTopology {

    private static Logger logger = Logger.getLogger(KingNetwork.class);
    protected Random random;
    protected int assignSpace;
    protected HashMap nodes;
    protected double interNodeLatencies[][];
    protected double intraNodeLatency;
    protected double interNodeLoss;
    protected double intraNodeLoss;
    protected double interNodeBandwidth;
    protected double intraNodeBandwidth;
    protected double latencyScaleFactor;
    protected double latencyScaleConstant;
    protected double nextLatency;
    protected double nextLoss;
    protected double nextBandwidth;
    protected InetAddress nextHop;

    /**
     * Constructor KingNetwork
     *
     * @param random
     * @param filename
     * @param intraNodeLatency
     * @param interNodeLoss
     * @param intraNodeLoss
     * @param interNodeBandwidth
     * @param intraNodeBandwidth
     * @param latencyScalePercentage
     * @param latencyScaleConstant
     */
    public KingNetwork(Random random, String filename, double intraNodeLatency,
                       double interNodeLoss, double intraNodeLoss,
                       double interNodeBandwidth, double intraNodeBandwidth,
                       double latencyScalePercentage,
                       double latencyScaleConstant) {
        this.random = random;
        this.intraNodeLatency =
            (intraNodeLatency * latencyScalePercentage / 100)
            + latencyScaleConstant;
        this.interNodeLoss = interNodeLoss;
        this.intraNodeLoss = intraNodeLoss;
        this.interNodeBandwidth = interNodeBandwidth;
        this.intraNodeBandwidth = intraNodeBandwidth;
        this.latencyScaleFactor = latencyScalePercentage / 100;
        this.latencyScaleConstant = latencyScaleConstant;
        this.assignSpace = readFile(filename);
        this.nodes = new HashMap();
    }

    /**
     * Method parseFile
     *
     * @param filename
     * @return
     */
    protected int readFile(String filename) {
        int lineno = 1;
        int nodes = 0;
        long totalLatency = 0;

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(this, "Started Parsing",
                                                  new Object[]{"f",
                                                               filename}, null));
        }

        try {
            BufferedReader reader =
                new BufferedReader(new FileReader(filename));
            StreamTokenizer tok = new StreamTokenizer(reader);

            tok.resetSyntax();
            tok.whitespaceChars((int) ',', (int) ',');
            tok.whitespaceChars((int) ' ', (int) ' ');
            tok.wordChars((int) '_', (int) '_');
            tok.wordChars((int) 'A', (int) 'Z');
            tok.wordChars((int) 'a', (int) 'z');
            tok.wordChars((int) '0', (int) '9');

            while (lineno < 5) {
                tok.nextToken();

                if (tok.ttype == StreamTokenizer.TT_EOL) {
                    lineno++;
                }
            }

            while (true) {
                tok.nextToken();

                if ((tok.ttype != StreamTokenizer.TT_WORD)
                        || ( !tok.sval.equals("node"))) {
                    break;
                }

                tok.nextToken();

                if (tok.ttype != StreamTokenizer.TT_WORD) {
                    break;
                }

                int n = Integer.parseInt(tok.sval);

                if (n != nodes + 1) {
                    throw new RuntimeException("Skipped a node number: " + n
                                               + ", line " + lineno);
                }

                tok.nextToken();

                if (tok.ttype != StreamTokenizer.TT_EOL) {
                    throw new RuntimeException("Expected a new line, line "
                                               + lineno);
                }

                ++lineno;
                ++nodes;
            }

            interNodeLatencies = new double[nodes + 1][];

            for (int i = 1; i < nodes + 1; ++i) {
                interNodeLatencies[i] = new double[nodes + 1];
            }

            for (int i = 1; i < nodes + 1; ++i) {
                for (int j = i + 1; j < nodes + 1; ++j) {
                    if (tok.ttype != StreamTokenizer.TT_WORD) {
                        throw new RuntimeException("Expected a number, line "
                                                   + lineno);
                    }

                    if (Integer.parseInt(tok.sval) != i) {
                        throw new RuntimeException(
                            "Expected first number to be " + i + ", line "
                            + lineno);
                    }

                    /*
                     * tok.nextToken ();
                     * if (tok.ttype != StreamTokenizer.TT_WORD) {
                     *   logger.fatal ("expected a comma " + lineno);
                     *   System.exit (1);
                     * }
                     */
                    tok.nextToken();

                    if (tok.ttype != StreamTokenizer.TT_WORD) {
                        throw new RuntimeException("Expected a number, line "
                                                   + lineno);
                    }

                    if (Integer.parseInt(tok.sval) != j) {
                        throw new RuntimeException(
                            "Expected second number to be " + j + ", line "
                            + lineno);
                    }

                    tok.nextToken();

                    if (tok.ttype != StreamTokenizer.TT_WORD) {
                        throw new RuntimeException("Expected a latency, line "
                                                   + lineno);
                    }

                    interNodeLatencies[i][j] = interNodeLatencies[j][i] =
                        (((double) Integer.parseInt(
                            tok.sval)) / 1000
                                       * latencyScaleFactor) + latencyScaleConstant;
                    totalLatency += interNodeLatencies[i][j];

                    tok.nextToken();

                    if (tok.ttype != StreamTokenizer.TT_EOL) {
                        throw new RuntimeException("Expected a new line, line "
                                                   + lineno);
                    }

                    ++lineno;

                    tok.nextToken();
                }
            }
        } catch (Exception exception) {
            throw new RuntimeException(exception);
        }

        double averageLatency = ((double) totalLatency)
                                / ((double) nodes * nodes * 1000 / 2);

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(this, "Finished Parsing",
                                                  new Object[]{"n",
                                                               String.valueOf(
                                                               nodes),
                                                               "l",
                                                               String.valueOf(
                                                               lineno),
                                                               "a",
                                                               String.valueOf(
                                                               averageLatency)}, null));
        }

        return nodes;
    }

    /**
     * Method linkNodeToTopology
     *
     * @param node
     * @param last
     */
    public void linkNodeToTopology(InetAddress node, boolean last) {
        int position = random.nextInt(assignSpace) + 1;

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(this,
                                                  "Mapping Node to Topology",
                                                  new Object[]{"i",
                                                               node,
                                                               "p",
                                                               String.valueOf(
                                                               position)}, null));
        }

        nodes.put(node, new Integer(position));

        if (last) {
            computeActualAverageLatency();
        }
    }

    /**
     * Method computeActualAverageLatency
     */
    protected void computeActualAverageLatency() {
        double diameter = 0;
        long interLatencyCount = 0;
        double interLatencySum = 0;
        long intraLatencyCount = 0;
        Iterator source = nodes.entrySet().iterator();

        while (source.hasNext()) {
            Map.Entry sourceNode = (Map.Entry) source.next();
            Iterator destination = nodes.entrySet().iterator();

            while (destination.hasNext()) {
                Map.Entry destinationNode = (Map.Entry) destination.next();

                if ( !(sourceNode.getKey().equals(destinationNode.getKey()))) {
                    int sourcePos =
                        ((Integer) sourceNode.getValue()).intValue();
                    int destinationPos =
                        ((Integer) destinationNode.getValue()).intValue();

                    if (sourcePos == destinationPos) {
                        intraLatencyCount++;
                    } else {
                        interLatencyCount++;

                        interLatencySum +=
                            interNodeLatencies[sourcePos][destinationPos];

                        if (interNodeLatencies[sourcePos][destinationPos]
                                > diameter) {
                            diameter =
                                interNodeLatencies[sourcePos][destinationPos];
                        }
                    }
                }
            }
        }

        double interLatencyAverage = interLatencySum
                                     / ((double) interLatencyCount);
        double totalAverage =
            (interLatencySum + (intraNodeLatency * intraLatencyCount))
            / ((double) (interLatencyCount + intraLatencyCount));

        logger.info(new StructuredLogMessage(this, "King Network Statistics",
                                             new Object[] {
            "n", String.valueOf(nodes.size()), "p",
            String.valueOf(interLatencyAverage), "s",
            String.valueOf(interLatencySum), "e",
            String.valueOf(interLatencyCount), "i",
            String.valueOf(intraLatencyCount), "a",
            String.valueOf(totalAverage), "d", String.valueOf(diameter)
        }, null, true));
    }

    /**
     * Method computeNextHop
     *
     * @param source
     * @param destination
     */
    public void computeNextHop(InetAddress source, InetAddress destination) {
        Integer pos1 = (Integer) nodes.get(source);

        if (pos1 == null) {
            throw new RuntimeException("Source node " + source
                                       + " not initialized in topology");
        }

        Integer pos2 = (Integer) nodes.get(destination);

        if (pos2 == null) {
            throw new RuntimeException("Destination node " + destination
                                       + " not initialized in topology");
        }

        int node1 = pos1.intValue();
        int node2 = pos2.intValue();

        if (node1 == node2) {
            nextLatency = intraNodeLatency;
            nextLoss = intraNodeLoss;
            nextBandwidth = intraNodeBandwidth;
        } else {
            nextLatency = interNodeLatencies[node1][node2];
            nextLoss = interNodeLoss;
            nextBandwidth = interNodeBandwidth;
        }

        nextHop = destination;
    }

    /**
     * Method getNextHop
     * @return
     */
    public InetAddress getNextHop() {
        return nextHop;
    }

    /**
     * Method getLatency
     * @return
     */
    public double getLatency() {
        return nextLatency;
    }

    /**
     * Method getLoss
     * @return
     */
    public double getLoss() {
        return nextLoss;
    }

    /**
     * Method getMaxBandwidth
     * @return
     */
    public double getMaxBandwidth() {
        return nextBandwidth;
    }
}