can.java

High performance DB query

            SerializationManager.getPayloadSize(route)
            - SerializationManager.getPayloadSize(introduce));
        LocalNode.myUDPMessenger.send(mySocketAddress, landmarkNode, route);
    }

    /**
     * Receive and handle an Introduce message. If accepting join, compute split of zone space
     * @param source     address of the node trying to join
     * @param introduce  the complete messsage containing the unique ID
     */
    private void processIntroduce(InetAddress source, Introduce introduce) {

        // Check if there is pending reorg, if so reject Introduce
        if ((active == true) && (joinID == -1) && (leaveID == -1)) {

            // Record the address and ID of the joining node
            joinID = introduce.getID();
            pendingNewcomer = introduce.getSource();

            // Create a new zone for the newcomer by splitting the next split dimension.  Keep the zone around, since we'll have its complement here if the newcomer acknowledges the split. I'll take the higher span of the split and give the lower span to the newcomer.
            double[][] intervals = zone.getIntervals();
            double[][] newIntervals = otherTempIntervals;
            double[][] myNewIntervals = myTempIntervals;

            for (int i = 0; i < DIM; i++) {
                newIntervals[i][0] = intervals[i][0];
                newIntervals[i][1] = intervals[i][1];
                myNewIntervals[i][0] = intervals[i][0];
                myNewIntervals[i][1] = intervals[i][1];
            }

            newIntervals[nextSplit][1] =
                (intervals[nextSplit][0] + intervals[nextSplit][1]) / 2.0;
            myNewIntervals[nextSplit][0] = newIntervals[nextSplit][1];
            pendingNewcomerZone = new Zone(newIntervals);    // store the split zone
            myPendingZone = new Zone(myNewIntervals);

            Neighborhood pendingNewNeighborhood = new Neighborhood(neighborhood,
                                                      zone,
                                                      pendingNewcomerZone,
                                                      mySocketAddress,
                                                      myPendingZone, nextSplit,
                                                      true);

            // Send welcome message
            if (Output.debuggingEnabled) {
                logger.debug(new LogMessage(new Object[]{
                    "Sending WELCOME message to ",
                    source, " with area ",
                    pendingNewcomerZone.toString(),
                    " leaving ",
                    myPendingZone}));
            }

            sendWelcome(introduce, pendingNewcomerZone, pendingNewNeighborhood);
        } else {
            sendRefuse(introduce);
        }
    }

    /**
     * This method sends a WELCOME message to an aspiring new CAN node
     * @param introduce
     * @param newZone
     * @param newNeighborhood
     */
    private void sendWelcome(Introduce introduce, Zone newZone,
                             Neighborhood newNeighborhood) {
        Welcome welcome = Welcome.allocate(introduce.getID(), mySocketAddress,
                                           newZone, (nextSplit + 1) % DIM,
                                           newNeighborhood);

        StatCollector.addSample(StatVars.NETWORK_OUT,
                                StatVars.LOCATION_SERVICE,
                                StatVars.CAN_WELCOME,
                                SerializationManager.getPayloadSize(welcome));
        LocalNode.myUDPMessenger.send(mySocketAddress, introduce.getSource(),
                                      welcome);
    }

    /**
     * This method sends a REFUSE message to an aspiring new CAN node
     * @param introduce
     */
    private void sendRefuse(Introduce introduce) {
        Refuse refuse = Refuse.allocate(introduce.getID(), mySocketAddress);

        StatCollector.addSample(StatVars.NETWORK_OUT,
                                StatVars.LOCATION_SERVICE, StatVars.CAN_REFUSE,
                                SerializationManager.getPayloadSize(refuse));
        LocalNode.myUDPMessenger.send(mySocketAddress, introduce.getSource(),
                                      refuse);
    }

    /** Receive and handle a WELCOME message.  If I am still waiting for a response to my INTRODUCE message, I accept the supplied Zone and Neighborhood and respond with an ACCEPT message.  Otherwise, I do nothing. */
    private void processWelcome(Welcome welcome) {

        // Check to make sure this is the right welcome message
        if (joinID == welcome.getID()) {
            active = true;

            // Install the zone and neighborhood
            zone = welcome.getZone();

            version++;

            nextSplit = welcome.getNextSplit();
            neighborhood = new Neighborhood(welcome.getNeighborhood(),
                                            LocalNode.myTimer.getCurrentTime());

            StatCollector.addSample(StatVars.MISC_B, StatVars.LOCATION_SERVICE,
                                    StatVars.ID_SPACE, zone.getArea());

            // Update new neighbors and schedule a cleanup
            updateNeighbors(neighborhood);
            LocalNode.myTimer.schedule(
                LocalNode.myRandom.nextDouble() * (MAX_UPDATE - MIN_UPDATE)
                + MIN_UPDATE, UPDATE_SIGNAL, this);
            LocalNode.myTimer.schedule(CLEANUP, CLEANUP_SIGNAL, this);

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

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

                client.locationMapChange();
            }

            // Accept the honor
            joinID = -1;

            if (Output.debuggingEnabled) {
                logger.debug(new LogMessage(new Object[]{
                    "Sending ACCEPT message to ",
                    welcome.getSource(),
                    " and has area ", zone}));
            }

            sendAccept(welcome.getID(), welcome.getSource());
        }
    }

    /** This method sends an ACCEPT message to the previous owner of my new zone. */
    private void sendAccept(int joinID, InetSocketAddress owner) {
        Accept accept = Accept.allocate(joinID, owner);

        StatCollector.addSample(StatVars.NETWORK_OUT,
                                StatVars.LOCATION_SERVICE, StatVars.CAN_ACCEPT,
                                SerializationManager.getPayloadSize(accept));
        LocalNode.myUDPMessenger.send(mySocketAddress, owner, accept);
    }

    /** Receive and handle an ACCEPT message.  If I am still waiting for an acceptance from a newly welcomed node, then install my retained half of the zone and move on.  Otherwise, ignore the message. */
    private void processAccept(InetAddress source, Accept accept) {

        // Check for the correct join ID from the proper address
        if ((joinID == accept.getID()) && (pendingNewcomer != null)
                && (pendingNewcomer.getAddress().equals(source))) {

            // Change to the smaller zone, install the associated neighborhood
            zone = myPendingZone;

            StatCollector.addSample(StatVars.MISC_B, StatVars.LOCATION_SERVICE,
                                    StatVars.ID_SPACE, zone.getArea());

            version++;

            Neighborhood oldNeighborhood = neighborhood;

            neighborhood = new Neighborhood(neighborhood, zone, myPendingZone,
                                            pendingNewcomer,
                                            pendingNewcomerZone, nextSplit,
                                            false);    // I'm keeping the high split
            nextSplit = (nextSplit + 1) % DIM;
            pendingNewcomer = null;
            joinID = -1;

            // Update all my old neighbors
            updateNeighbors(oldNeighborhood);

            if (Output.debuggingEnabled) {
                logger.debug(new LogMessage(new Object[]{
                    "Join complete, ready for additional joins/leaves"}));
            }
        } else if (Output.debuggingEnabled) {
            logger.debug(new LogMessage(new Object[]{
                "Received bad ACCEPT message from ",
                source}));
        }
    }

    /** Receive and handle a Route message.  Check if the coordinates belong to me.  If so, decapsulate the message and handle it. Otherwise pass it on to the next hop towards its destination. */
    private void processRoute(InetAddress source, Route route) {
        if (Output.debuggingEnabled) {
            logger.debug(new LogMessage(new Object[]{
                "Processing ROUTE message: ",
                String.valueOf(route.getID()),
                " from: ", source}));
        }

        // Do the coordinates belong to my zone?
        if (zone.contains(route.getCoordinates())) {

            // only increment hop count if I didn't send it.
            if (source.getAddress().equals(mySocketAddress.getAddress())
                    == false) {
                route.incrementHopCount(1);
            }

            // charge stats for cost of ROUTE overhead, the remaining net bandwidth will be charged in doPayload against embedded messasge type
            StatCollector.addSample(
                StatVars.NETWORK_IN, StatVars.LOCATION_SERVICE,
                StatVars.CAN_ROUTE,
                SerializationManager.getPayloadSize(route)
                - SerializationManager.getPayloadSize(route.getPayload()));
            StatCollector.addSample(StatVars.MISC_A, StatVars.LOCATION_SERVICE,
                                    StatVars.HOP_COUNT, route.getHopCount());

            // decapsulate and resubmit to doPayload
            handleUDPNetwork(route.getSource(), route.getPayload());
            Route.free(route);
        } else {

            // Route message not processed further, so charge stats completely for routing
            StatCollector.addSample(StatVars.NETWORK_IN,
                                    StatVars.LOCATION_SERVICE,
                                    StatVars.CAN_ROUTE,
                                    SerializationManager.getPayloadSize(route));

            // Process UpCalls
            boolean upCall = false;

            if (route.getProvideUpCalls()
                    && (route.getPayload() instanceof Message)) {
                Message theMessage = (Message) route.getPayload();

                upCall = handleMessageUpCall(
                    theMessage.getDestination(), theMessage,
                    source.getAddress().equals(mySocketAddress.getAddress()));
            }

            if ( !upCall) {

                // Forward to the next-hop neighbor in the neighborhood
                route.incrementHopCount(1);
                sendForward(route,
                            neighborhood.nextHop(route.getCoordinates()));
            }
        }
    }

    private boolean handleMessageUpCall(BitID destination, Message message,
                                        boolean local) {
        LocationServiceClient client =
            (LocationServiceClient) applicationsCallback.get(
                new Long(message.getApplicationID()));

        if (client != null) {
            return client.routeUpCall(destination, message.getMessage(), local);
        }

        return false;
    }

    /** Receive and handle a REFUSE message.  If I am waiting for a WELCOME from the owner of my chosen zone, then wait for a bit, and try again, picking a different zone.  Otherwise, ignore the message. */
    private void processRefuse(Refuse refuse) {

        // Am I waiting for this joinID?
        if ((joinID != -1) && (joinID == refuse.getID())) {
            double sleepyTime = LocalNode.myRandom.nextDouble()
                                * (MAX_RETRY - MIN_RETRY) + MIN_RETRY;

            LocalNode.myTimer.schedule(sleepyTime, RETRY_SIGNAL, this);

            if (Output.debuggingEnabled) {
                logger.debug(new LogMessage(new Object[]{
                    "Denied admission, waiting for ",
                    String.valueOf(sleepyTime),
                    " before retrying."}));
            }
        }
    }

    /**
     * Lookup
     *
     * @param source
     * @param lookup
     */
    public void processLookup(InetAddress source, Lookup lookup) {
        LookupResponse response = LookupResponse.allocate(
                                      lookup.getID(), mySocketAddress,
                                      (InetSocketAddress) applications.get(
                                          new Long(lookup.getApplicationID())));

        StatCollector.addSample(StatVars.NETWORK_OUT,
                                StatVars.LOCATION_SERVICE,
                                StatVars.CAN_LOOKUPRESPONSE,
                                SerializationManager.getPayloadSize(response));
        LocalNode.myUDPMessenger.send(
            mySocketAddress,
            new InetSocketAddress(source, portNumber.intValue()), response);
    }

    /** Receive and handle an UPDATE message.  Go through my neighbors. If I find the updating neighbor, mark it with the current time and update its zone. If the new zone is no longer my neighbor, eliminate the neighbor.  If this isn't a current neighbor, check if its zone is neighboring.  If so, include the sender to my appropriate neighborhood.  Otherwise ignore the message. */
    private void processUpdate(Update update) {
        neighborhood.update(zone, update.getSource(), update.getZone(),
                            update.getVersion(),
                            LocalNode.myTimer.getCurrentTime());

        if (update.getDepth() > 0) {
            relayDeepUpdate(update, neighborhood);
        }
    }

    /** Relay deep UPDATE to all my neighbors, except for the one who sent this in the first place. */
    private void relayDeepUpdate(Update update, Neighborhood neighborhood) {