simudpmessenger.java

High performance DB query

/*
 * @(#)$Id: SimUDPMessenger.java,v 1.10 2004/12/15 02:20:07 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.services.network.udp;

import services.network.Payload;
import services.network.udp.UDPClient;
import services.network.udp.UDPwithAckClient;
import services.network.udp.UDPwithAckMessenger;
import services.stats.StatCollector;
import services.stats.StatVars;
import simulator.schedulers.clock.ClockStub;
import simulator.schedulers.network.IPMessage;
import simulator.schedulers.network.NetworkClient;
import simulator.schedulers.network.NetworkStub;
import util.network.serialization.GenericByteBuffer;
import util.network.serialization.SerializationManager;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashMap;

/**
 * Class SimUDPMessenger
 *
 */
public class SimUDPMessenger implements UDPwithAckMessenger, NetworkClient {

    private NetworkStub networkStub;
    private InetAddress myIP;
    private Integer outPortNum;
    private ClockStub theClockStub;
    private int latencyHistory;
    private HashMap ports;
    private HashMap latencyTables;

    /**
     * Constructor SimUDPMessenger
     *
     * @param networkStub
     * @param myIP
     * @param outPortNum
     * @param theClockStub
     * @param latencyHistory
     */
    public SimUDPMessenger(NetworkStub networkStub, InetAddress myIP,
                           Integer outPortNum, ClockStub theClockStub,
                           int latencyHistory) {
        this.networkStub = networkStub;
        this.myIP = myIP;
        this.outPortNum = outPortNum;
        this.theClockStub = theClockStub;
        this.latencyHistory = latencyHistory;
        ports = new HashMap();
        latencyTables = new HashMap();
    }

    /**
     * Method send
     *
     * @param destination
     * @param payload
     */
    public void send(InetSocketAddress destination, Payload payload) {
        send(null, destination, payload, null, null);
    }

    /**
     * Method send
     *
     * @param source
     * @param destination
     * @param payload
     */
    public void send(InetSocketAddress source, InetSocketAddress destination,
                     Payload payload) {
        send(source, destination, payload, null, null);
    }

    /**
     * Method send
     *
     * @param destination
     * @param payload
     * @param client
     * @param ackData
     */
    public void send(InetSocketAddress destination, Payload payload,
                     UDPwithAckClient client, Object ackData) {
        send(null, destination, payload, client, ackData);
    }

    /**
     * Method getRoundTripLatency
     *
     * @param destination
     * @return
     */
    public long getRoundTripLatency(InetSocketAddress destination) {
        LatencyEntry entry = (LatencyEntry) latencyTables.get(destination);

        if (entry != null) {
            return entry.getAverage();
        } else {
            return 0;
        }
    }

    /**
     * Method getRoundTripLatency
     *
     * @param source
     * @param destination
     * @return
     */
    public long getRoundTripLatency(InetSocketAddress source,
                                    InetSocketAddress destination) {
        return getRoundTripLatency(destination);
    }

    /**
     * Method send
     *
     * @param source
     * @param destination
     * @param payload
     * @param client
     * @param ackData
     */
    public void send(InetSocketAddress source, InetSocketAddress destination,
                     Payload payload, UDPwithAckClient client, Object ackData) {
        AckObject ackObject = null;

        if (client != null) {
            ackObject = new AckObject(client, ackData);
        }

        SimUDPDataMessage message = null;

        if (source != null) {
            message =
                SimUDPDataMessage.allocate(new Integer(source.getPort()),
                                           new Integer(destination.getPort()),
                                           payload, ackObject,
                                           theClockStub.getCurrentTime());
        } else {
            message =
                SimUDPDataMessage.allocate(outPortNum,
                                           new Integer(destination.getPort()),
                                           payload, ackObject,
                                           theClockStub.getCurrentTime());
        }

        if (destination.getAddress().equals(myIP)) {
            StatCollector.addSample(
                StatVars.NETWORK_OUT, StatVars.UDP_NETWORK,
                StatVars.UDP_DATA_INTERNAL,
                SerializationManager.getPayloadSize(message));
        } else {
            StatCollector.addSample(
                StatVars.NETWORK_OUT, StatVars.UDP_NETWORK, StatVars.UDP_DATA,
                SerializationManager.getPayloadSize(message));
        }

        networkStub.send(destination.getAddress(), IPMessage.PROTOCOL_UDP,
                         message);
    }

    /**
     * Class AckObject
     *
     */
    public class AckObject implements Payload {

        public UDPwithAckClient client;
        public Object ackData;

        /**
         * Method serialize
         *
         * @param outputBuffer
         * @return
         */
        public long serialize(GenericByteBuffer outputBuffer) {
            return 0;
        }

        /**
         * Constructor AckObject
         *
         * @param client
         * @param ackData
         */
        public AckObject(UDPwithAckClient client, Object ackData) {
            this.client = client;
            this.ackData = ackData;
        }

        /**
         * Method getSize
         * @return
         */
        public int getSize() {
            return 0;
        }
    }


    /**
     * Method listen
     *
     * @param portNumber
     * @param client
     * @return
     */
    public boolean listen(Integer portNumber, UDPClient client) {
        if (ports.containsValue(portNumber)) {
            return false;
        }

        ports.put(portNumber, client);

        return true;
    }

    /**
     * Method release
     *
     * @param portNumber
     */
    public void release(Integer portNumber) {
        ports.remove(portNumber);
    }

    private void doUDPDataMessage(InetSocketAddress source,
                                  SimUDPDataMessage message) {
        UDPClient client =
            (UDPClient) ports.get(message.getDestinationPortNumber());

        if (client != null) {
            client.handleUDPNetwork(source, message.getData());
        }
    }

    /**
     * Method handleNetwork
     *
     * @param source
     * @param data
     */
    public void handleNetwork(InetAddress source, Payload data) {
        if (data instanceof SimUDPDataMessage) {
            if (source.equals(myIP)) {
                StatCollector.addSample(
                    StatVars.NETWORK_IN, StatVars.UDP_NETWORK,
                    StatVars.UDP_DATA_INTERNAL,
                    SerializationManager.getPayloadSize(data));
            } else {
                StatCollector.addSample(
                    StatVars.NETWORK_IN, StatVars.UDP_NETWORK,
                    StatVars.UDP_DATA,
                    SerializationManager.getPayloadSize(data));
            }

            SimUDPDataMessage message = (SimUDPDataMessage) data;
            InetSocketAddress sourceSocket =
                new InetSocketAddress(source,
                                      message.getSourcePortNumber().intValue());
            AckObject ackObject = (AckObject) message.getAckData();

            if (ackObject != null) {
                networkStub.send(
                    sourceSocket.getAddress(), IPMessage.PROTOCOL_UDP,
                    SimUDPAckMessage.allocate(
                        ackObject, message.getSendTime()));
            }

            doUDPDataMessage(sourceSocket, message);
            SimUDPDataMessage.free(message);
        }

        if (data instanceof SimUDPAckMessage) {
            SimUDPAckMessage message = (SimUDPAckMessage) data;

            updateLatencyTable(
                source,
                (long) ((theClockStub.getCurrentTime() - message.getSendTime())
                        * 1000));

            AckObject ackObject = (AckObject) message.getData();

            if (ackObject != null) {
                ackObject.client.handleUDPNetworkAck(ackObject.ackData, true);
            }
        }
    }

    private void updateLatencyTable(InetAddress destination,
                                    long roundTripLatency) {
        LatencyEntry entry = (LatencyEntry) latencyTables.get(destination);

        if (entry == null) {
            entry = new LatencyEntry();

            latencyTables.put(destination, entry);
        }

        entry.addReading(roundTripLatency);
    }

    private class LatencyEntry {

        public int position;
        public long[] readings;
        private boolean full;

        /**
         * Constructor LatencyEntry
         */
        public LatencyEntry() {
            position = 0;
            readings = new long[latencyHistory];
            full = false;
        }

        /**
         * Method addReading
         *
         * @param newReading
         */
        public void addReading(long newReading) {
            readings[position] = newReading;

            position++;

            if (position >= latencyHistory) {
                position = 0;
                full = true;
            }
        }

        /**
         * Method getAverage
         * @return
         */
        public long getAverage() {
            int maxReadings = (full)
                              ? (latencyHistory - 1)
                              : position;
            long total = 0;

            for (int i = 0; i < maxReadings; i++) {
                total += readings[i];
            }

            return total / (maxReadings + 1);
        }
    }
}