basicprovider.java

High performance DB query

/*
 * @(#)$Id: BasicProvider.java,v 1.51 2005/10/10 00:59:41 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 overlay.provider.basic;

import java.math.BigInteger;
import java.net.InetSocketAddress;
import java.util.*;
import org.apache.log4j.Logger;
import overlay.location.LocationService;
import overlay.location.LocationServiceClient;
import overlay.provider.Provider;
import overlay.provider.ProviderClient;
import overlay.storage.StorageManager;
import services.LocalNode;
import services.Output;
import services.network.Payload;
import services.network.udp.UDPClient;
import services.stats.StatCollector;
import services.stats.StatVars;
import services.timer.TimerClient;
import util.BitID;
import util.logging.LogMessage;
import util.logging.StructuredLogMessage;
import util.network.serialization.SerializationManager;

/**
 * Class BasicProvider
 *
 */
public class BasicProvider
        implements Provider, LocationServiceClient, UDPClient, TimerClient {

    private static Logger logger = Logger.getLogger(BasicProvider.class);
    private static final Integer SIGNAL_CLEANUP = new Integer(0);
    private static final char nsSizeDelimiter = '/';
    private static final char nsIgnoreChar = '*';
    public static char[] lastChar = {Character.MAX_VALUE};
    public static String lastCharString = new String(lastChar);
    private LocationService locationService;
    private StorageManager storageManager;
    private InetSocketAddress listeningSocketAddress;
    private long cleanupPeriod;
    private long applicationID;
    private boolean removeImmediately;
    private boolean rehashStorageOnChange;
    private int idBits;
    private int defaultNSBits;
    private String delimiter;
    private boolean storeUpCalls;
    private int messageID;
    private HashMap requestClients;
    private HashMap waitingRequests;
    private TreeMap newDataCallbacks, messageCallbacks, messageUpCallCallbacks;

    // Maintains a mapping of item's common id (ns + rid + iid) to storage ID. Only actual stored items should be the directory. Used for LSCAN
    private TreeMap storageDirectory;

    /**
     * Constructor BasicProvider
     *
     * @param locationService
     * @param storageManager
     * @param applicationID
     * @param cleanupPeriod
     * @param providerPort
     * @param removeImmediately
     * @param rehashStorageOnChange
     * @param idBits
     * @param defaultNSBits
     * @param delimiter
     * @param storeUpCalls
     */
    public BasicProvider(LocationService locationService,
                         StorageManager storageManager, long applicationID,
                         long cleanupPeriod, int providerPort,
                         boolean removeImmediately,
                         boolean rehashStorageOnChange, int idBits,
                         int defaultNSBits, String delimiter,
                         boolean storeUpCalls) {

        // install the services
        this.locationService = locationService;
        this.storageManager = storageManager;
        this.applicationID = applicationID;
        this.cleanupPeriod = cleanupPeriod;

        // listen on provider port
        SerializationManager.registerClass("overlay.provider.basic.GetMessage");
        SerializationManager.registerClass(
            "overlay.provider.basic.GetResponseMessage");
        SerializationManager.registerClass(
            "overlay.provider.basic.ProviderMessage");
        SerializationManager.registerClass("overlay.provider.basic.PutMessage");
        SerializationManager.registerClass(
            "overlay.provider.basic.PutResponseMessage");
        SerializationManager.registerClass(
            "overlay.provider.basic.RenewMessage");
        SerializationManager.registerClass(
            "overlay.provider.basic.RenewResponseMessage");
        SerializationManager.registerClass(
            "overlay.provider.basic.SendMessage");
        SerializationManager.registerClass("overlay.provider.basic.StoredItem");
        SerializationManager.registerClass(
            "overlay.provider.basic.StoredItemCollection");

        listeningSocketAddress = new InetSocketAddress(LocalNode.myIPAddress,
                                                       providerPort);

        if (LocalNode.myUDPMessenger.listen(new Integer(providerPort), this)
                == false) {
            throw new RuntimeException("Duplicate use of port number "
                                       + providerPort);
        }

        // flag if items should be kept till expiration even if no longer local
        this.removeImmediately = removeImmediately;
        this.rehashStorageOnChange = rehashStorageOnChange;
        this.idBits = idBits;
        this.defaultNSBits = defaultNSBits;
        this.delimiter = delimiter;
        this.storeUpCalls = storeUpCalls;

        // initialize data structures
        messageID = 0;
        requestClients = new HashMap();
        waitingRequests = new HashMap();
        newDataCallbacks = new TreeMap();
        messageCallbacks = new TreeMap();
        messageUpCallCallbacks = new TreeMap();
        storageDirectory = new TreeMap();

        // register for location map changes
        locationService.registerClient(this, applicationID,
                                       listeningSocketAddress);

        // Schedule first cleanup
        LocalNode.myTimer.schedule((cleanupPeriod / 1000), SIGNAL_CLEANUP,
                                   this);
    }

    /**
     * Method get
     *
     * @param ns
     * @param rid
     * @param client
     */
    public void get(String ns, String rid, ProviderClient client) {
        BitID locationID = computeLocationID(ns, rid);

        get(null, locationID, ns, rid, client);
    }

    /**
     * Method get
     *
     * @param locationNS
     * @param locationRID
     * @param ns
     * @param rid
     * @param client
     */
    public void get(String locationNS, String locationRID, String ns,
                    String rid, ProviderClient client) {
        BitID locationID = computeLocationID(locationNS, locationRID);

        get(null, locationID, ns, rid, client);
    }

    /**
     * Method get
     *
     * @param locationID
     * @param ns
     * @param rid
     * @param client
     */
    public void get(BitID locationID, String ns, String rid,
                    ProviderClient client) {
        get(null, locationID, ns, rid, client);
    }

    /**
     * Method get
     *
     * @param location
     * @param locationID
     * @param ns
     * @param rid
     * @param client
     */
    public void get(InetSocketAddress location, BitID locationID, String ns,
                    String rid, ProviderClient client) {
        Integer requestID = new Integer(messageID++);
        GetMessage theMessage = GetMessage.allocate(requestID, ns, rid,
                                                    locationID);

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(theMessage, "Get Request",
                                                  new Object[] {
                "h", location, "d", locationID, "n", ns, "r", rid, "m",
                requestID, "c", client
            }, null));
        }

        if (location == null) {
            issueLookup(locationID, requestID, client, theMessage);
        } else {
            sendMessage(location, theMessage, StatVars.PROVIDER_PUT);
        }
    }

    /**
     * Method put
     *
     * @param ns
     * @param rid
     * @param iid
     * @param item
     * @param lifetime
     * @param client
     */
    public void put(String ns, String rid, int iid, Payload item,
                    long lifetime, ProviderClient client) {
        BitID locationID = computeLocationID(ns, rid);

        put(null, locationID, ns, rid, iid, item, lifetime, client);
    }

    /**
     * Method put
     *
     * @param locationNS
     * @param locationRID
     * @param ns
     * @param rid
     * @param iid
     * @param item
     * @param lifetime
     * @param client
     */
    public void put(String locationNS, String locationRID, String ns,
                    String rid, int iid, Payload item, long lifetime,
                    ProviderClient client) {
        BitID locationID = computeLocationID(locationNS, locationRID);

        put(null, locationID, ns, rid, iid, item, lifetime, client);
    }

    /**
     * Method put
     *
     *
     * @param locationID
     * @param ns
     * @param rid
     * @param iid
     * @param item
     * @param lifetime
     * @param client
     */
    public void put(BitID locationID, String ns, String rid, int iid,
                    Payload item, long lifetime, ProviderClient client) {
        put(null, locationID, ns, rid, iid, item, lifetime, client);
    }

    /**
     * Method put
     *
     * @param location
     * @param locationID
     * @param ns
     * @param rid
     * @param iid
     * @param item
     * @param lifetime
     * @param client
     */
    public void put(InetSocketAddress location, BitID locationID, String ns,
                    String rid, int iid, Payload item, long lifetime,
                    ProviderClient client) {
        Integer requestID = new Integer(messageID++);
        PutMessage theMessage = PutMessage.allocate(requestID, ns, rid,
                                                    locationID, iid, lifetime,
                                                    item);

        if (Output.debuggingEnabled) {
            logger.debug(new StructuredLogMessage(theMessage, "Put Request",
                                                  new Object[] {
                "h", location, "d", locationID, "n", ns, "r", rid, "i",
                String.valueOf(iid), "l", String.valueOf(lifetime), "m",
                requestID, "c", client, "t",
                Integer.toHexString(item.hashCode())
            }, new Object[]{"p", item}));
        }

        if (location == null) {
            issueLookup(locationID, requestID, client, theMessage);
        } else {
            sendMessage(location, theMessage, StatVars.PROVIDER_PUT);
        }
    }

    /**
     * Method send
     *
     * @param ns
     * @param rid
     * @param iid
     * @param item
     * @param lifetime
     * @param provideUpCalls
     * @param sendLocal
     */
    public void send(String ns, String rid, int iid, Payload item,
                     long lifetime, boolean provideUpCalls, boolean sendLocal) {
        BitID locationID = computeLocationID(ns, rid);

        send(null, locationID, ns, rid, iid, item, lifetime, provideUpCalls,
             sendLocal);
    }

    /**
     * Method send
     *
     * @param locationNS
     * @param locationRID
     * @param ns
     * @param rid
     * @param iid
     * @param item
     * @param lifetime
     * @param provideUpCalls
     * @param sendLocal
     */
    public void send(String locationNS, String locationRID, String ns,
                     String rid, int iid, Payload item, long lifetime,
                     boolean provideUpCalls, boolean sendLocal) {
        BitID locationID = computeLocationID(locationNS, locationRID);

        send(null, locationID, ns, rid, iid, item, lifetime, provideUpCalls,
             sendLocal);
    }

    /**
     * Method send
     *
     * @param locationID
     * @param ns
     * @param rid
     * @param iid
     * @param item
     * @param lifetime
     * @param provideUpCalls
     * @param sendLocal
     */
    public void send(BitID locationID, String ns, String rid, int iid,
                     Payload item, long lifetime, boolean provideUpCalls,
                     boolean sendLocal) {
        send(null, locationID, ns, rid, iid, item, lifetime, provideUpCalls,
             sendLocal);
    }

    /**
     * Method send
     *
     *
     *
     * @param location
     * @param locationID
     * @param ns
     * @param rid