fairnetworkscheduler.java

High performance DB query

/*
 * @(#)$Id: FairNetworkScheduler.java,v 1.6 2004/07/02 23:59:22 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.fair;

import java.net.InetAddress;
import org.apache.log4j.Logger;
import services.Output;
import simulator.core.Event;
import simulator.core.NodeMap;
import simulator.core.Simulator;
import simulator.schedulers.network.IPMessage;
import simulator.schedulers.network.NetworkEvent;
import simulator.schedulers.network.NetworkTopology;
import simulator.schedulers.network.node.naive.NaiveNetworkScheduler;
import util.logging.LogMessage;
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 FairNetworkScheduler extends NaiveNetworkScheduler {

    private static Logger logger = Logger.getLogger(FairNetworkScheduler.class);

    /**
     * Constructor FairNetworkScheduler
     *
     * @param theSimulator
     * @param theTopology
     * @param theNodes
     */
    public FairNetworkScheduler(Simulator theSimulator,
                                NetworkTopology theTopology, NodeMap theNodes) {
        super(theSimulator, theTopology, theNodes);
    }

    /**
     * Method processForDelivery
     *
     * @param theEvent
     */
    protected void processForDelivery(FairNetworkEvent theEvent) {
        IPMessage theMessage = theEvent.getMessage();
        InetAddress currentHop = theEvent.getImmediateDestination();
        FairNetworkStub destinationStub =
            (FairNetworkStub) getNetworkStub(currentHop);

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

        Flow theFlow = destinationStub.deregisterFlow(Flow.DESTINATION,
                                                      theEvent);

        ((FairNetworkStub) theFlow.getStub(Flow.SOURCE)).deregisterFlow(
            Flow.SOURCE, theEvent);

        double latency = theTopology.getLatency();
        double curTime = getCurrentTime();
        FairNetworkEvent theNewEvent = FairNetworkEvent.allocate(this,
                                           theEvent.getCreationTime(),
                                           curTime + latency,
                                           theEvent.getNodeNum(), theMessage,
                                           theEvent.getImmediateDestination(),
                                           true);

        theSimulator.schedule(theNewEvent);
    }

    /**
     * Method handleEvent
     *
     * @param theEvent
     */
    public void handleEvent(Event theEvent) {
        FairNetworkEvent theFairEvent = (FairNetworkEvent) theEvent;

        if (theFairEvent.getFinalDelivery()) {
            processNetworkEvent(theFairEvent);
        } else {
            processForDelivery(theFairEvent);
        }
    }

    /**
     * Method scheduleMessage
     *
     * @param theMessage
     * @param currentHop
     */
    protected void scheduleMessage(IPMessage theMessage,
                                   InetAddress currentHop) {
        FairNetworkStub sourceStub =
            (FairNetworkStub) 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();
        FairNetworkStub destinationStub =
            (FairNetworkStub) getNetworkStub(nextHop);

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

        double sourceBandwidth =
            sourceStub.getNewFlowBandwidth(FairNetworkStub.OUTGOING);
        double destinationBandwidth =
            destinationStub.getNewFlowBandwidth(FairNetworkStub.INCOMING);
        double bandwidth =
            Math.min(Math.min(sourceBandwidth, destinationBandwidth),
                     theTopology.getMaxBandwidth());
        double transmissionTime =
            SerializationManager.getPayloadSize(theMessage) / bandwidth;
        double curTime = getCurrentTime();
        FairNetworkEvent theEvent = FairNetworkEvent.allocate(this, curTime,
                                        curTime + transmissionTime,
                                        getNodeNum(nextHop), theMessage,
                                        nextHop, false);
        Flow theFlow = Flow.allocate(theEvent, theMessage, sourceStub,
                                     destinationStub, sourceBandwidth,
                                     destinationBandwidth, curTime);

        sourceStub.registerFlow(Flow.SOURCE, theEvent, theFlow);
        destinationStub.registerFlow(Flow.DESTINATION, theEvent, theFlow);
        theSimulator.schedule(theEvent);

        if (Output.debuggingEnabled) {
            logger.debug(new LogMessage(new Object[]{
                "Scheduling Message from: ",
                theMessage.getSource(),
                " to ",
                theMessage.getDestination(),
                "(", currentHop, " to ",
                nextHop, ")"}));
        }
    }

    /**
     * Method rescheduleFlow
     *
     * @param theFlow
     * @param timeRemaining
     */
    protected void rescheduleFlow(Flow theFlow, double timeRemaining) {
        NetworkEvent theOldEvent = theFlow.getNetworkEvent();

        theSimulator.deschedule(theOldEvent);

        FairNetworkEvent theNewEvent =
            FairNetworkEvent.allocate(
                this, theOldEvent.getCreationTime(),
                theSimulator.getCurrentTime() + timeRemaining,
                theOldEvent.getNodeNum(), theOldEvent.getMessage(),
                theOldEvent.getImmediateDestination(), false);

        ((FairNetworkStub) theFlow.getStub(Flow.SOURCE)).updateRegisteredFlow(
            Flow.SOURCE, theOldEvent, theNewEvent);
        ((FairNetworkStub) theFlow.getStub(
            Flow.DESTINATION)).updateRegisteredFlow(
                Flow.DESTINATION, theOldEvent, theNewEvent);
        theFlow.resetNetworkEvent(theNewEvent);
        NetworkEvent.free(theOldEvent);
        theSimulator.schedule(theNewEvent);
    }
}