simtcpmessenger.java

High performance DB query

/*
 * @(#)$Id: SimTCPMessenger.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.services.network.tcp;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashMap;
import services.network.Payload;
import services.network.tcp.TCPClient;
import services.network.tcp.TCPConnection;
import services.network.tcp.TCPMessenger;
import services.stats.StatCollector;
import services.stats.StatVars;
import simulator.schedulers.network.IPMessage;
import simulator.schedulers.network.NetworkClient;
import simulator.schedulers.network.NetworkStub;
import util.network.serialization.SerializationManager;

/**
 * Class SimTCPMessenger
 *
 */
public class SimTCPMessenger implements TCPMessenger, NetworkClient {

    private static final int TCP_MAX_PORT = 65536;
    private NetworkStub networkStub;
    private Integer minOutPortNum;
    private Integer curOutPortNum;
    private HashMap ports;
    private HashMap connections;

    /**
     * Constructor SimTCPMessenger
     *
     * @param networkStub
     * @param minOutPortNum
     */
    public SimTCPMessenger(NetworkStub networkStub, Integer minOutPortNum) {
        this.networkStub = networkStub;
        this.minOutPortNum = minOutPortNum;
        curOutPortNum = minOutPortNum;
        ports = new HashMap();
        connections = new HashMap();
    }

    /**
     * Method listen
     *
     * @param portNumber
     * @param client
     * @return
     */
    public boolean listen(Integer portNumber, TCPClient 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);
    }

    /**
     * Method connect
     *
     * @param destination
     * @param client
     * @return
     */
    public TCPConnection connect(InetSocketAddress destination,
                                 TCPClient client) {
        SimTCPConnection connection =
            (SimTCPConnection) connections.get(destination);

        if (connection == null) {
            Integer outPortNumber = nextOutPortNumber();

            connection =
                SimTCPConnection.allocate(destination.getAddress(),
                                          new Integer(destination.getPort()),
                                          outPortNumber, this, true);

            connections.put(destination, connection);
            listen(outPortNumber, client);
        }

        connect(connection);

        return connection;
    }

    /**
     * Method connect
     *
     * @param connection
     */
    protected void connect(TCPConnection connection) {
        if (connection.isConnected() || connection.isConnectionPending()) {
            ((SimTCPConnection) connection).open();

            SimTCPSYNMessage syn =
                SimTCPSYNMessage.allocate(
                    new Integer(connection.getRemoteSocketAddress().getPort()),
                    new Integer(connection.getLocalSocketAddress().getPort()));

            StatCollector.addSample(StatVars.NETWORK_OUT, StatVars.TCP_NETWORK,
                                    StatVars.TCP_SYN,
                                    SerializationManager.getPayloadSize(syn));
            networkStub.send(connection.getRemoteSocketAddress().getAddress(),
                             IPMessage.PROTOCOL_TCP, syn);
        }
    }

    /**
     * Method disconnect
     *
     * @param connection
     */
    public void disconnect(TCPConnection connection) {
        int unsentData = ((SimTCPConnection) connection).waitingData();

        if (connection.isConnected() || connection.isConnectionPending()) {
            SimTCPRSTMessage rst =
                SimTCPRSTMessage.allocate(
                    new Integer(connection.getRemoteSocketAddress().getPort()),
                    new Integer(connection.getLocalSocketAddress().getPort()));

            networkStub.send(connection.getRemoteSocketAddress().getAddress(),
                             IPMessage.PROTOCOL_TCP, rst);
        }

        ((SimTCPConnection) connection).closeConnection();

        if (((SimTCPConnection) connection).isOutbound()) {
            release(new Integer(connection.getLocalSocketAddress().getPort()));
        }

        SimTCPConnection.free(((SimTCPConnection) connection));
    }

    /**
     * Method send
     *
     * @param destination
     * @param client
     * @param data
     * @return
     */
    public TCPConnection send(InetSocketAddress destination, TCPClient client,
                              Payload data) {
        TCPConnection connection = connect(destination, client);

        send(connection, data);

        return connection;
    }

    /**
     * Method send
     *
     * @param connection
     * @param data
     * @return
     */
    public boolean send(TCPConnection connection, Payload data) {
        if (connection.isClosed()) {
            return false;
        }

        ((SimTCPConnection) connection).queueData(data);

        return true;
    }

    /**
     * Method networkSend
     *
     * @param destination
     * @param message
     */
    protected void networkSend(InetAddress destination,
                               SimTCPDataMessage message) {
        StatCollector.addSample(StatVars.NETWORK_OUT, StatVars.TCP_NETWORK,
                                StatVars.TCP_DATA,
                                SerializationManager.getPayloadSize(message));
        networkStub.send(destination, IPMessage.PROTOCOL_TCP, message);
    }

    private boolean isOpenPortNumber(Integer portNumber) {
        if (ports.get(portNumber) == null) {
            return true;
        } else {
            return false;
        }
    }

    private Integer nextOutPortNumber() {
        int curAttempt = curOutPortNum.intValue();
        int firstAttempt = curAttempt;

        while ( !isOpenPortNumber(curOutPortNum)) {
            curAttempt++;

            if (curAttempt > TCP_MAX_PORT) {
                curAttempt = minOutPortNum.intValue();
            }

            curOutPortNum = new Integer(curAttempt);

            if (curAttempt == firstAttempt) {
                throw new RuntimeException(
                    "TCPMesseneger: Out of out-bound port numbers");
            }
        }

        return curOutPortNum;
    }

    private void doTCPSYNMessage(InetAddress source, SimTCPSYNMessage message) {
        TCPClient client =
            (TCPClient) ports.get(message.getDestinationPortNumber());

        if (client != null) {
            SimTCPConnection connection =
                SimTCPConnection.allocate(source,
                                          message.getSourcePortNumber(),
                                          message.getDestinationPortNumber(),
                                          this, false);

            connections.put(
                new InetSocketAddress(
                    source,
                    message.getSourcePortNumber().intValue()), connection);
            connection.completeConnection();
            connection.sendNext();

            SimTCPSYNACKMessage synack =
                SimTCPSYNACKMessage.allocate(
                    message.getSourcePortNumber(),
                    message.getDestinationPortNumber());

            StatCollector.addSample(
                StatVars.NETWORK_OUT, StatVars.TCP_NETWORK,
                StatVars.TCP_SYNACK,
                SerializationManager.getPayloadSize(synack));
            networkStub.send(source, IPMessage.PROTOCOL_TCP, synack);
        } else {
            SimTCPRSTMessage rst =
                SimTCPRSTMessage.allocate(message.getSourcePortNumber(),
                                          message.getDestinationPortNumber());

            StatCollector.addSample(StatVars.NETWORK_OUT, StatVars.TCP_NETWORK,
                                    StatVars.TCP_RST,
                                    SerializationManager.getPayloadSize(rst));
            networkStub.send(source, IPMessage.PROTOCOL_TCP, rst);
        }
    }

    private void doTCPSYNACKMessage(InetAddress source,
                                    SimTCPSYNACKMessage message) {
        SimTCPConnection connection =
            (SimTCPConnection) connections.get(new InetSocketAddress(source,
                message.getSourcePortNumber().intValue()));

        if (connection != null) {
            connection.completeConnection();
            connection.sendNext();
        } else {
            SimTCPRSTMessage rst =
                SimTCPRSTMessage.allocate(message.getSourcePortNumber(),
                                          message.getDestinationPortNumber());

            StatCollector.addSample(StatVars.NETWORK_OUT, StatVars.TCP_NETWORK,
                                    StatVars.TCP_RST,
                                    SerializationManager.getPayloadSize(rst));
            networkStub.send(source, IPMessage.PROTOCOL_TCP, rst);
        }
    }

    private void doTCPRSTMessage(InetAddress source, SimTCPRSTMessage message) {
        SimTCPConnection connection =
            (SimTCPConnection) connections.get(new InetSocketAddress(source,
                message.getSourcePortNumber().intValue()));

        if (connection != null) {
            disconnect(connection);
        }
    }

    private void doTCPDataMessage(InetAddress source,
                                  SimTCPDataMessage message) {
        TCPClient client =
            (TCPClient) ports.get(message.getDestinationPortNumber());
        SimTCPConnection connection =
            (SimTCPConnection) connections.get(new InetSocketAddress(source,
                message.getSourcePortNumber().intValue()));

        if (client != null) {
            client.handleTCPNetwork(connection, message.getData());

            SimTCPACKMessage ack =
                SimTCPACKMessage.allocate(message.getSourcePortNumber(),
                                          message.getDestinationPortNumber(),
                                          message.getSequenceNumber());

            StatCollector.addSample(StatVars.NETWORK_OUT, StatVars.TCP_NETWORK,
                                    StatVars.TCP_ACK,
                                    SerializationManager.getPayloadSize(ack));
            networkStub.send(source, IPMessage.PROTOCOL_TCP, ack);
            message.getSourceConnection().sendNext();
        }
    }

    private void doTCPACKMessage(InetAddress source, SimTCPACKMessage message) {
        SimTCPConnection connection =
            (SimTCPConnection) connections.get(new InetSocketAddress(source,
                message.getSourcePortNumber().intValue()));

        if (connection != null) {
            connection.processACK(message.getSequenceNumber());
        }
    }

    /**
     * Method handleNetwork
     *
     * @param source
     * @param data
     */
    public void handleNetwork(InetAddress source, Payload data) {
        if (data instanceof SimTCPSYNMessage) {
            StatCollector.addSample(StatVars.NETWORK_IN, StatVars.TCP_NETWORK,
                                    StatVars.TCP_SYN,
                                    SerializationManager.getPayloadSize(data));

            SimTCPSYNMessage message = (SimTCPSYNMessage) data;

            doTCPSYNMessage(source, message);
            SimTCPSYNMessage.free(message);
        }

        if (data instanceof SimTCPSYNACKMessage) {
            StatCollector.addSample(StatVars.NETWORK_IN, StatVars.TCP_NETWORK,
                                    StatVars.TCP_SYNACK,
                                    SerializationManager.getPayloadSize(data));

            SimTCPSYNACKMessage message = (SimTCPSYNACKMessage) data;

            doTCPSYNACKMessage(source, message);
            SimTCPSYNACKMessage.free(message);
        }

        if (data instanceof SimTCPRSTMessage) {
            StatCollector.addSample(StatVars.NETWORK_IN, StatVars.TCP_NETWORK,
                                    StatVars.TCP_RST,
                                    SerializationManager.getPayloadSize(data));

            SimTCPRSTMessage message = (SimTCPRSTMessage) data;

            doTCPRSTMessage(source, message);
            SimTCPRSTMessage.free(message);
        }

        if (data instanceof SimTCPDataMessage) {
            StatCollector.addSample(StatVars.NETWORK_IN, StatVars.TCP_NETWORK,
                                    StatVars.TCP_DATA,
                                    SerializationManager.getPayloadSize(data));

            SimTCPDataMessage message = (SimTCPDataMessage) data;

            doTCPDataMessage(source, message);
            SimTCPDataMessage.free(message);
        }

        if (data instanceof SimTCPACKMessage) {
            StatCollector.addSample(StatVars.NETWORK_IN, StatVars.TCP_NETWORK,
                                    StatVars.TCP_ACK,
                                    SerializationManager.getPayloadSize(data));

            SimTCPACKMessage message = (SimTCPACKMessage) data;

            doTCPACKMessage(source, message);
            SimTCPACKMessage.free(message);
        }
    }
}