fifonetworkscheduler.java

High performance DB query

/*
 * @(#)$Id: FifoNetworkScheduler.java,v 1.6 2005/08/02 19:51:33 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.node.fifo;

import java.net.InetAddress;
import services.stats.StatCollector;
import services.stats.StatVars;
import simulator.core.Event;
import simulator.core.NodeMap;
import simulator.core.Simulator;
import simulator.schedulers.network.IPMessage;
import simulator.schedulers.network.NetworkTopology;
import simulator.schedulers.network.node.naive.NaiveNetworkScheduler;
import util.network.serialization.SerializationManager;

/**
 * Approximates fluid flow systems. This implementation has some major deficiencies...
 * 1) It is not work-conserving, each flow is allocated an equal share of the in/out bandwidth for a node whether or not it is used. This prevents the need for a change in one flow to potentially require rescheduling every other flow.
 * 2) Bandwidth is allocated from both nodes from the time the message begins to be sent till it is completely received, in reality the sending node should only be 'charged' for bandwidth for the sending, while the receiving node only for receving, neither should be charged for the latency.
 */
public class FifoNetworkScheduler extends NaiveNetworkScheduler {

    protected long statPeriod;

    /**
     * Constructor FifoNetworkScheduler
     *
     * @param theSimulator
     * @param theTopology
     * @param theNodes
     * @param statPeriod
     */
    public FifoNetworkScheduler(Simulator theSimulator,
                                NetworkTopology theTopology, NodeMap theNodes,
                                long statPeriod) {
        super(theSimulator, theTopology, theNodes);

        this.statPeriod = statPeriod;

        if (this.statPeriod > 0) {
            Event theEvent = new Event(this, 0, 0, 0);

            theSimulator.schedule(theEvent);
        }
    }

    /**
     * Method processForDelivery
     *
     * @param theEvent
     */
    protected void processForDelivery(FifoNetworkEvent theEvent) {
        IPMessage theMessage = theEvent.getMessage();
        FifoNetworkStub destinationStub =
            (FifoNetworkStub) getNetworkStub(theMessage.getDestination());

        if (destinationStub == null) {
            throw new RuntimeException(
                "Could not locate destination node to deliver message "
                + theMessage);
        }

        double curTime = getCurrentTime();
        double deliveryTime = destinationStub.deliveryTime(
                                  FifoNetworkStub.INCOMING, curTime,
                                  SerializationManager.getPayloadSize(
                                      theEvent.getMessage()));
        FifoNetworkEvent theNewEvent = FifoNetworkEvent.allocate(this,
                                           theEvent.getCreationTime(),
                                           deliveryTime, theEvent.getNodeNum(),
                                           theMessage,
                                           theEvent.getImmediateDestination(),
                                           true);

        theSimulator.schedule(theNewEvent);
    }

    /**
     * Method handleEvent
     *
     * @param theEvent
     */
    public void handleEvent(Event theEvent) {
        if (theEvent instanceof FifoNetworkEvent) {
            FifoNetworkEvent theFifoEvent = (FifoNetworkEvent) theEvent;

            if (theFifoEvent.getFinalDelivery()) {
                processNetworkEvent(theFifoEvent);
            } else {
                processForDelivery(theFifoEvent);
            }
        } else {
            collectStats();
        }
    }

    /**
     * Method scheduleMessage
     *
     * @param theMessage
     * @param currentHop
     */
    protected void scheduleMessage(IPMessage theMessage,
                                   InetAddress currentHop) {
        FifoNetworkStub sourceStub =
            (FifoNetworkStub) getNetworkStub(currentHop);

        if (sourceStub == null) {
            throw new RuntimeException(
                "Could not locate source node to send message " + theMessage);
        }

        theTopology.computeNextHop(currentHop, theMessage.getDestination());

        InetAddress nextHop = theTopology.getNextHop();
        FifoNetworkStub destinationStub =
            (FifoNetworkStub) getNetworkStub(nextHop);

        if (destinationStub == null) {
            throw new RuntimeException(
                "Could not locate next hop node to send message " + theMessage);
        }

        double latency = theTopology.getLatency();
        double curTime = getCurrentTime();
        double arrivalTime = latency
                             + sourceStub.deliveryTime(
                                 FifoNetworkStub.OUTGOING, curTime,
                                 SerializationManager.getPayloadSize(
                                     theMessage));
        FifoNetworkEvent theEvent = FifoNetworkEvent.allocate(this, curTime,
                                        arrivalTime, getNodeNum(nextHop),
                                        theMessage, nextHop, false);

        theSimulator.schedule(theEvent);
    }

    /**
     * Method collectStats
     */
    protected void collectStats() {

        // Collect stats
        double curTime = getCurrentTime();
        double[] delaySum = new double[2];
        double[] delayMax = new double[2];
        int[] delayMaxNode = new int[2];

        for (int i = 0; i < 2; i++) {
            delaySum[i] = 0;
            delayMax[i] = 0;
            delayMaxNode[i] = 0;
        }

        int numNodes = theNodes.numNodes();

        for (int j = 0; j < numNodes; j++) {
            FifoNetworkStub stub =
                (FifoNetworkStub) theNodes.getNode(j).getNetworkStub();

            for (int i = 0; i < 2; i++) {
                double delay = stub.getQueueDelay(i, curTime);

                delaySum[i] += delay;

                if (delay > delayMax[i]) {
                    delayMax[i] = delay;
                    delayMaxNode[i] = j;
                }
            }
        }

        // Record stats
        for (int i = 0; i < 2; i++) {
            StatEntry entry = new StatEntry(curTime, delaySum[i], delayMax[i],
                                            delayMaxNode[i]);
            int statgroup = (i == 0)
                            ? StatVars.QUEUE_OUT_LENGTH
                            : StatVars.QUEUE_IN_LENGTH;

            StatCollector.addSample(StatVars.MISC_C, StatVars.IP_NETWORK,
                                    statgroup, entry);
        }

        // Schedule next stat collections
        double nextTime = curTime + (statPeriod / 1000);
        Event theEvent = new Event(this, curTime, nextTime, 0);

        theSimulator.schedule(theEvent);
    }

    /**
     * Class StatEntry
     *
     */
    public class StatEntry {

        public double time;
        public double sum;
        public double max;
        public int maxNode;

        /**
         * Constructor StatEntry
         *
         * @param time
         * @param sum
         * @param max
         * @param maxNode
         */
        public StatEntry(double time, double sum, double max, int maxNode) {
            this.time = time;
            this.sum = sum;
            this.max = max;
            this.maxNode = maxNode;
        }

        /**
         * Method toString
         * @return
         */
        public String toString() {
            return time + "," + sum + "," + max + "," + maxNode;
        }
    }
}