bamboo.java

High performance DB query

/*
 * @(#)$Id: Bamboo.java,v 1.23 2005/05/28 20:13:05 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.
 */


/* IMPORTANT: READ BEFORE DOWNLOADING, COPYING, INSTALLING OR USING.  By downloading, copying, installing or using the software you agree to this license.  If you do not agree to this license, do not download, install, copy or use the software.
 * == Berkeley License ==
 * Copyright (c) 2003 Regents of the University of California.  All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
 * 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
 * 3. Neither the name of the University nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS 'AS IS' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * == Intel Open Source License ==
 * Copyright (c) 2003 Intel Corporation  All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
 *        Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
 *    Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
 *    Neither the name of the Intel Corporation nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE INTEL OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package overlay.location.bamboo;

import java.io.Reader;
import java.io.StringReader;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import org.apache.log4j.Logger;
import overlay.location.LocationService;
import overlay.location.LocationServiceClient;
import overlay.location.bamboo.payload.Lookup;
import overlay.location.bamboo.payload.LookupResponse;
import overlay.location.bamboo.payload.Message;
import services.LocalNode;
import services.Output;
import services.network.Payload;
import services.network.udp.UDPClient;
import services.stats.StatCollector;
import services.stats.StatVars;
import util.BitID;
import util.logging.LogMessage;
import util.logging.StructuredLogMessage;
import util.network.serialization.SerializationManager;
import util.timer.timeouts.TimeoutManager;
import util.timer.timeouts.TimeoutManagerClient;

/**
 * Class Bamboo
 *
 */
public class Bamboo
        implements LocationService, TimeoutManagerClient, UDPClient {

    private static Logger logger = Logger.getLogger(Bamboo.class);
    public static final int BAMBOO_BYTE_OVERHEAD =
        24 + 24 + 8 + 1 + 8
        + 8;    // src + dest + appID + other + PayloadEmulatorOverhead
    private int port, dustDevilPort;
    private double timeout;
    private boolean iterative;
    private boolean ignoreProximity;
    private int guidBase;
    private int leafSetSize;
    private int locationCacheSize;
    private String explicitGuid;
    private DustDevilEmulator bambooRuntime;
    private BambooSubsystem bambooSubsystem;
    private ArrayList localClients;
    private HashMap lookupRequests, lookupRequestsID, applications,
                    applicationsCallback;
    private InetSocketAddress address, dustDevilAddress;
    private TimeoutManager timeoutManager;
    private int messageID;

    /**
     * Constructor Bamboo
     *
     * @param port
     * @param dustDevilPort
     * @param timeout
     * @param iterative
     * @param ignoreProximity
     * @param guidBase
     * @param leafSetSize
     * @param locationCacheSize
     */
    public Bamboo(int port, int dustDevilPort, double timeout,
                  boolean iterative, boolean ignoreProximity, int guidBase,
                  int leafSetSize, int locationCacheSize) {
        this.port = dustDevilPort;
        this.dustDevilPort = port;
        this.timeout = timeout;
        this.iterative = iterative;
        this.ignoreProximity = ignoreProximity;
        this.guidBase = guidBase;
        this.leafSetSize = leafSetSize;
        this.locationCacheSize = locationCacheSize;
        this.address = new InetSocketAddress(LocalNode.myIPAddress, this.port);
        this.dustDevilAddress = new InetSocketAddress(LocalNode.myIPAddress,
                                                      this.dustDevilPort);
        this.timeoutManager = new TimeoutManager();
        this.messageID = 0;

        init();

        try {
            Class programClass =
                Class.forName("overlay.location.bamboo.DustDevilEmulatorImpl");

            bambooRuntime = (DustDevilEmulator) programClass.newInstance();

            Class subsystemClass =
                Class.forName("overlay.location.bamboo.BambooSubsystemImpl");

            bambooSubsystem = (BambooSubsystem) subsystemClass.newInstance();
        } catch (Exception exception) {
            throw new RuntimeException(
                "Unable to load Bamboo/DustDevil Runtime: "
                + exception.getClass() + " " + exception.getMessage());
        }
    }

    private void init() {
        localClients = new ArrayList();
        lookupRequests = new HashMap();
        lookupRequestsID = new HashMap();
        applications = new HashMap();
        applicationsCallback = new HashMap();
    }

    /**
     * Method lookup
     *
     * @param locationID
     * @param applicationID
     * @param requestor
     * @param requestID
     */
    public void lookup(BitID locationID, long applicationID,
                       LocationServiceClient requestor, Object requestID) {
        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(requestID, "Lookup Request",
                                                  new Object[]{"m",
                                                               String.valueOf(
                                                               messageID),
                                                               "d",
                                                               locationID,
                                                               "i",
                                                               requestID,
                                                               "c",
                                                               requestor,
                                                               "a",
                                                               String.valueOf(
                                                               applicationID)}, null));
        }

        if (requestID == null) {
            return;
        }

        Integer localMap = new Integer(messageID);

        lookupRequests.put(localMap, requestor);
        lookupRequestsID.put(localMap, requestID);

        Lookup request = Lookup.allocate(messageID++, address, locationID,
                                         applicationID);

        if (timeout > 0) {
            timeoutManager.addTimeout(request, localMap, timeout, this);
        }

        StatCollector.addSample(StatVars.NETWORK_OUT,
                                StatVars.LOCATION_SERVICE,
                                StatVars.BAMBOO_LOOKUP,
                                SerializationManager.getPayloadSize(request)
                                + BAMBOO_BYTE_OVERHEAD);
        bambooSubsystem.sendMessage(locationID, applicationID, true, iterative,
                                    request);
    }

    /**
     * Method send
     *
     * @param locationID
     * @param applicationID
     * @param message
     * @param provideUpCalls
     */
    public void send(BitID locationID, long applicationID, Payload message,
                     boolean provideUpCalls) {
        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(message, "Send Request",
                                                  new Object[]{"m",
                                                               String.valueOf(
                                                               messageID),
                                                               "d",
                                                               locationID,
                                                               "u",
                                                               String.valueOf(
                                                               provideUpCalls),
                                                               "a",
                                                               String.valueOf(
                                                               applicationID)}, new Object[]{
                                                                   "p",
                                                                   message, }));
        }

        Message request = Message.allocate(messageID++, address, locationID,
                                           applicationID, message,
                                           provideUpCalls);

        StatCollector.addSample(StatVars.NETWORK_OUT,
                                StatVars.LOCATION_SERVICE,
                                StatVars.BAMBOO_MESSAGE,
                                SerializationManager.getPayloadSize(request)
                                + BAMBOO_BYTE_OVERHEAD);
        bambooSubsystem.sendMessage(locationID, applicationID, true, iterative,
                                    request);
    }

    /**
     * Method getLocationID
     * @return
     */
    public BitID getLocationID() {
        return bambooSubsystem.getLocationID();
    }

    /**
     * Method setGuid
     *
     * @param explicitGuid
     */
    public void setGuid(String explicitGuid) {
        this.explicitGuid = explicitGuid;
    }

    /**
     * Method join
     *
     * @param landmarkNode
     */
    public void join(InetSocketAddress landmarkNode) {
        join(new InetSocketAddress[]{landmarkNode});
    }

    /**
     * Method join
     *
     * @param landmarkNodes
     */
    public void join(InetSocketAddress[] landmarkNodes) {
        SerializationManager.registerClass(
            "overlay.location.bamboo.payload.BambooMessage");
        SerializationManager.registerClass(
            "overlay.location.bamboo.payload.Lookup");
        SerializationManager.registerClass(
            "overlay.location.bamboo.payload.LookupResponse");
        SerializationManager.registerClass(
            "overlay.location.bamboo.payload.Message");

        String dustDevilHostName = dustDevilAddress.getAddress().toString();

        dustDevilHostName =
            dustDevilHostName.substring(dustDevilHostName.indexOf('/') + 1);

        String globalConfig = "<global>\n <initargs>\n node_id "
                              + dustDevilHostName + ":"
                              + dustDevilAddress.getPort()
                              + "\n </initargs>\n </global>\n";
        String networkStageConfig =
            "<Network>\n class bamboo.network.Network\n <initargs>\n debug_level 0\n </initargs>\n </Network>\n";

        if ((landmarkNodes == null) || (landmarkNodes.length == 0)
                || (landmarkNodes[0] == null)) {
            landmarkNodes = new InetSocketAddress[]{dustDevilAddress};
        }

        int numGateways = landmarkNodes.length;
        String routerStageConfig =
            "<Router-" + dustDevilHostName
            + ">\n class bamboo.router.Router\n <initargs>\n gateway_count "
            + numGateways + "\n";

        for (int i = 0; i < numGateways; i++) {
            String landmarkNodeHostName =
                landmarkNodes[i].getAddress().toString();

            landmarkNodeHostName =
                landmarkNodeHostName.substring(landmarkNodeHostName.indexOf('/')
                                               + 1);
            routerStageConfig += " gateway_" + i + " " + landmarkNodeHostName
                                 + ":" + landmarkNodes[i].getPort() + "\n";
        }

        routerStageConfig += "immediate_join true\n";

        if (ignoreProximity == true) {
            routerStageConfig += "ignore_proximity true\n";
        } else {
            routerStageConfig += "ignore_proximity false\n";
        }

        if (guidBase > 0) {
            routerStageConfig += "digit_values " + guidBase + "\n";
        }

        if (explicitGuid != null) {
            routerStageConfig += "explicit_guid " + explicitGuid + "\n";
        }

        if (leafSetSize >= 0) {
            routerStageConfig += "leaf_set_size " + leafSetSize + "\n";
        }

        if (locationCacheSize >= 0) {