chord.java

来自「High performance DB query」· Java 代码 · 共 1,010 行 · 第 1/3 页

/*
 * @(#)$Id: Chord.java,v 1.19 2004/12/31 19:53:43 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.location.chord;

import org.apache.log4j.Logger;
import overlay.location.LocationService;
import overlay.location.LocationServiceClient;
import overlay.location.chord.payload.*;
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.network.serialization.SerializationManager;
import util.timer.timeouts.TimeoutManager;
import util.timer.timeouts.TimeoutManagerClient;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

/* The main chord file */

/**
 * Class Chord
 *
 */
public class Chord
        implements LocationService, TimerClient, UDPClient,
                   TimeoutManagerClient {

    private static Logger logger = Logger.getLogger(Chord.class);
    private static final Integer SIGNAL_STABILIZE = new Integer(1);
    private static final Integer SIGNAL_MAINTAIN = new Integer(2);
    private boolean doRouteMaintenance = true;
    private int port, numBits, maxCachedFingers, maxHops;
    private boolean recursive;
    private double stabilizeDelay, maintainDelay, fingerLifetime,
                   cachedFingerLifetime, defaultLatency, latencyWeight, timeout;
    private BitID id;
    private FingerTable normalFingers, cachedFingers;
    private ArrayList localClients;
    private HashMap lookupRequests, lookupRequestsID, applications,
                    applicationsCallback;
    private InetSocketAddress address;
    private TimeoutManager timeoutManager;
    private InetSocketAddress landmarkNode;
    private int messageID;

    /**
     * Constructor Chord
     *
     * @param port
     * @param recursive
     * @param numBits
     * @param maxCachedFingers
     * @param stabilizeDelay
     * @param maintainDelay
     * @param fingerLifetime
     * @param cachedFingerLifetime
     * @param defaultLatency
     * @param latencyWeight
     * @param maxHops
     * @param timeout
     */
    public Chord(int port, boolean recursive, int numBits,
                 int maxCachedFingers, double stabilizeDelay,
                 double maintainDelay, double fingerLifetime,
                 double cachedFingerLifetime, double defaultLatency,
                 double latencyWeight, int maxHops, double timeout) {
        this.port = port;
        this.recursive = recursive;
        this.numBits = numBits;
        this.maxCachedFingers = maxCachedFingers;
        this.stabilizeDelay = stabilizeDelay;
        this.maintainDelay = maintainDelay;
        this.fingerLifetime = fingerLifetime;    // idLength * maintainDelay*2;
        this.cachedFingerLifetime = cachedFingerLifetime;
        this.defaultLatency = defaultLatency;
        this.latencyWeight = latencyWeight;
        this.maxHops = maxHops;
        this.timeout = timeout;
        this.messageID = 0;
        this.applications = new HashMap();
        this.applicationsCallback = new HashMap();
        this.address = new InetSocketAddress(LocalNode.myIPAddress, this.port);
        this.defaultLatency = defaultLatency;
        this.timeoutManager = new TimeoutManager();
        this.id = generateRandomID();

        // Initialize data structures
        init();
    }

    private BitID generateRandomID() {
        BitID theID = new BitID();

        for (int i = 0; i < numBits; i++) {
            if (LocalNode.myRandom.nextBoolean()) {
                theID.set(i, true);
            }
        }

        if (Output.debuggingEnabled) {
            logger.debug(new LogMessage(new Object[]{"Random Id: ",
                                                     theID.toNumString()}));
        }

        return theID;
    }

    private void init() {
        normalFingers = new FingerTable(id, address, numBits * 2,
                                        fingerLifetime, latencyWeight);
        cachedFingers = new FingerTable(id, address, maxCachedFingers,
                                        cachedFingerLifetime, latencyWeight);
        localClients = new ArrayList();
        lookupRequests = new HashMap();
        lookupRequestsID = new HashMap();
    }

    /**
     * Method reset
     */
    public void reset() {
        init();

        applications = new HashMap();
        applicationsCallback = new HashMap();
    }

    /* Network is being created artificially */

    /**
     * Method setRouteMaintenance
     *
     * @param doRouteMaintenance
     */
    public void setRouteMaintenance(boolean doRouteMaintenance) {
        this.doRouteMaintenance = doRouteMaintenance;
    }

    /**
     * Method fakeJoinStart
     *
     * @param id
     * @return
     */
    public Object[] fakeJoinStart(BitID id) {
        this.id = id;

        init();

        Object[] returnData = new Object[2];

        returnData[0] = normalFingers;
        returnData[1] = address;

        return returnData;
    }

    /**
     * Method join
     *
     * @param landmarkNode
     */
    public void join(InetSocketAddress landmarkNode) {
        SerializationManager.registerClass(
            "overlay.location.chord.payload.ChordMessage");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.Join");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.JoinRefuse");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.JoinResponse");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.Lookup");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.LookupResponse");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.Maintain");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.MaintainResponse");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.Message");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.Notify");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.Route");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.RouteResponse");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.Stabilize");
        SerializationManager.registerClass(
            "overlay.location.chord.payload.StabilizeResponse");

        if ( !(LocalNode.myUDPMessenger.listen(new Integer(port), this))) {
            throw new RuntimeException("Chord: Can't bind to port: " + port);
        }

        if (landmarkNode != null) {
            this.landmarkNode = landmarkNode;

            sendJoinRequest();
        } else {

            // No other nodes, setup stabilze and finger maintaince routines
            stabilize();
            maintainFingers();
        }
    }

    /**
     * Method join
     *
     * @param landmarkNodes
     */
    public void join(InetSocketAddress[] landmarkNodes) {
        if (landmarkNodes.length > 0) {
            join(landmarkNodes[0]);
        }
    }

    /**
     * Method leave
     */
    public void leave() {}

    /**
     * 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 LogMessage(new Object[]{
                "Issuing LOOKUP request ID: ",
                String.valueOf(messageID),
                ", FOR:",
                locationID.toNumString(),
                ", LOCALREQUEST:",
                requestID}));
        }

        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.CHORD_LOOKUP,
                                SerializationManager.getPayloadSize(request));
        routeMessage(request, locationID, true, false);
    }

    /**
     * Send a message to a node responsible for a locationID value
     * @param locationID the identifier of interest
     * @param applicationID the application intended to supply answer
     * @param message the message that will be hopped along the path to the destination
     * @param provideUpCalls true if upcalls should be attempted along path, false if just destination should get called
     */
    public void send(BitID locationID, long applicationID, Payload message,
                     boolean provideUpCalls) {
        if (Output.debuggingEnabled) {
            logger.debug(new LogMessage(new Object[]{
                "Issuing MESSAGE request ID: ",
                String.valueOf(messageID),
                ", FOR:",
                locationID.toNumString()}));
        }

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

        StatCollector.addSample(StatVars.NETWORK_OUT,
                                StatVars.LOCATION_SERVICE,
                                StatVars.CHORD_MESSAGE,
                                SerializationManager.getPayloadSize(request));
        routeMessage(request, locationID, true, provideUpCalls);
    }

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

    private void notifyMappingChange() {

        // after graceful leave call all global clients that leave has occured
        Iterator clientIterator = localClients.iterator();

        while (clientIterator.hasNext()) {
            LocationServiceClient client =
                (LocationServiceClient) clientIterator.next();