org/objectweb/proactive/p2p/service/P2PService.java

/* 
 * ################################################################
 * 
 * ProActive: The Java(TM) library for Parallel, Distributed, 
 *            Concurrent computing with Security and Mobility
 * 
 * Copyright (C) 1997-2007 INRIA/University of Nice-Sophia Antipolis
 * Contact: proactive@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *  
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *  
 *  Initial developer(s):               The ProActive Team
 *                        http://www.inria.fr/oasis/ProActive/contacts.html
 *  Contributor(s): 
 * 
 * ################################################################
 */ 
package org.objectweb.proactive.p2p.service;

import java.io.IOException;
import java.io.Serializable;
import java.util.Random;
import java.util.UUID;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.Body;
import org.objectweb.proactive.InitActive;
import org.objectweb.proactive.ProActive;
import org.objectweb.proactive.ProActiveInternalObject;
import org.objectweb.proactive.Service;
import org.objectweb.proactive.core.Constants;
import org.objectweb.proactive.core.body.UniversalBody;
import org.objectweb.proactive.core.body.request.Request;
import org.objectweb.proactive.core.body.request.RequestFilter;
import org.objectweb.proactive.core.mop.MOP;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.core.node.NodeFactory;
import org.objectweb.proactive.core.runtime.ProActiveRuntimeImpl;
import org.objectweb.proactive.core.util.log.Loggers;
import org.objectweb.proactive.core.util.log.ProActiveLogger;
import org.objectweb.proactive.core.util.wrapper.IntWrapper;
import org.objectweb.proactive.core.util.wrapper.StringWrapper;
import org.objectweb.proactive.p2p.service.exception.P2POldMessageException;
import org.objectweb.proactive.p2p.service.node.P2PNode;
import org.objectweb.proactive.p2p.service.node.P2PNodeLookup;
import org.objectweb.proactive.p2p.service.node.P2PNodeManager;
import org.objectweb.proactive.p2p.service.util.P2PConstants;


public class P2PService implements InitActive, P2PConstants, Serializable,
    ProActiveInternalObject {

    private static final Logger logger = ProActiveLogger.getLogger(Loggers.P2P_SERVICE);

    private P2PService acquaintances;

    private P2PAcquaintanceManager acquaintanceManager;

    private Node p2pServiceNode = null;
    private static final int MSG_MEMORY = Integer.parseInt(System.getProperty(
                P2PConstants.PROPERTY_MSG_MEMORY));
    private static final int NOA = Integer.parseInt(System.getProperty(
                P2PConstants.PROPERTY_NOA));
    private static final int EXPL_MSG = Integer.parseInt(System.getProperty(
                P2PConstants.PROPERTY_EXPLORING_MSG)) - 1;
    private static final long ACQ_TO = Long.parseLong(System.getProperty(
                P2PConstants.PROPERTY_NODES_ACQUISITION_T0));

    private static final Random randomizer = new Random();

    private Vector<UUID> oldMessageList = new Vector<UUID>(MSG_MEMORY);
    private P2PNodeManager nodeManager = null;

    private Vector<P2PNodeLookup> waitingNodesLookup = new Vector<P2PNodeLookup>();
    private Vector<P2PNodeLookup> waitingMaximunNodesLookup = new Vector<P2PNodeLookup>();
    private P2PService stubOnThis = null;

    // For asking nodes
    private Service service = null;
    private RequestFilter filter = new RequestFilter() {

            public boolean acceptRequest(Request request) {
                String requestName = request.getMethodName();
                if (requestName.compareToIgnoreCase("askingNode") == 0) {
                    return false;
                }
                return true;
            }
        };

    //--------------------------------------------------------------------------
    // Class Constructors
    //--------------------------------------------------------------------------

    public P2PService() {
        // empty
    }

    //--------------------------------------------------------------------------
    // Public Class methods
    // -------------------------------------------------------------------------

    public void firstContact(Vector peers) {
        // Creating an active P2PFirstContact
        Object[] params = new Object[3];
        params[0] = peers;
        params[1] = this.acquaintanceManager;
        params[2] = this.stubOnThis;
        try {
            ProActive.newActive(P2PFirstContact.class.getName(), params,
                this.p2pServiceNode);
        } catch (ActiveObjectCreationException e) {
            logger.warn("Couldn't active P2PFirstContact", e);
        } catch (NodeException e) {
            logger.warn("Couldn't active P2PFirstContact", e);
        }
    }

    public void register(P2PService service) {
        try {
            if (!this.stubOnThis.equals(service)) {
                this.acquaintanceManager.add(service);
                if (logger.isDebugEnabled()) {
                    logger.debug("Remote peer localy registered: " +
                        ProActive.getActiveObjectNodeUrl(service));
                }

                // Wake up all node accessor, because new peers are know
                this.wakeUpEveryBody();
            }
        } catch (Exception e) {
            logger.debug("The remote P2P service is certainly down", e);
        }
    }

    public void heartBeat() {
        logger.debug("Heart-beat message received");
    }

    public void exploring(int ttl, UUID uuid, P2PService remoteService) {
        if (uuid != null) {
            logger.debug("Exploring message received with #" + uuid);
            ttl--;
        }

        boolean broadcast;
        try {
            broadcast = broadcaster(ttl, uuid, remoteService);
        } catch (P2POldMessageException e) {
            return;
        }

        // This should be register
        if (this.shouldBeAcquaintance(remoteService)) {
            this.register(remoteService);
            try {
                remoteService.register(this.stubOnThis);
            } catch (Exception e) {
                logger.debug("Trouble with registering remote peer", e);
                this.acquaintanceManager.remove(remoteService);
            }
        } else if (broadcast) {
            // Forwarding the message
            if (uuid == null) {
                logger.debug("Generating uuid for exploring message");
                uuid = UUID.randomUUID();
            }
            this.acquaintances.exploring(ttl, uuid, remoteService);
            logger.debug("Broadcast exploring message with #" + uuid);
            uuid = null;
        }
    }

    public void askingNode(int ttl, UUID uuid, P2PService remoteService,
        int numberOfNodes, P2PNodeLookup lookup, String vnName, String jobId,
        String nodeFamilyRegexp) {
        boolean broadcast;
        if (uuid != null) {
            logger.debug("AskingNode message received with #" + uuid);
            ttl--;
            try {
                broadcast = broadcaster(ttl, uuid, remoteService);
            } catch (P2POldMessageException e) {
                return;
            }
        } else {
            broadcast = true;
        }

        // Do not give a local node to a local request
        if ((uuid != null) || (numberOfNodes == MAX_NODE)) {
            if (numberOfNodes == MAX_NODE) {
                Vector nodes = this.nodeManager.askingAllNodes(nodeFamilyRegexp);
                for (int i = 0; i < nodes.size(); i++) {
                    Node current = (Node) nodes.get(i);
                    if (vnName != null) {
                        try {
                            current.getProActiveRuntime()
                                   .registerVirtualNode(vnName, true);
                        } catch (Exception e) {
                            logger.warn("Couldn't register " + vnName +
                                " in the PAR", e);
                        }
                    }
                    if (jobId != null) {
                        current.getNodeInformation().setJobID(jobId);
                    }
                }
                if (nodes.size() > 0) {
                    lookup.giveNodeForMax(nodes, this.nodeManager);
                }
            } else {
                P2PNode askedNode = this.nodeManager.askingNode(nodeFamilyRegexp);

                // Asking node available?
                Node nodeAvailable = askedNode.getNode();
                if (nodeAvailable != null) {
                    IntWrapper nodeAck;

                    try {
                        nodeAck = lookup.giveNode(nodeAvailable,
                                askedNode.getNodeManager());
                    } catch (Exception lookupExcption) {
                        logger.info("Cannot contact the remote lookup",
                            lookupExcption);
                        this.nodeManager.noMoreNodeNeeded(nodeAvailable);
                        return;
                    }
                    if (nodeAck != null) {
                        // Waitng the ACK
                        long endTime = System.currentTimeMillis() + ACQ_TO;
                        while ((System.currentTimeMillis() < endTime) &&
                                ProActive.isAwaited(nodeAck)) {
                            if (this.service.hasRequestToServe()) {
                                service.serveAll(this.filter);
                            } else {
                                try {
                                    Thread.sleep(100);
                                } catch (InterruptedException e) {
                                    logger.debug(e);
                                }
                            }
                        }

                        // Testing future is here or timeout is expired??
                        if (ProActive.isAwaited(nodeAck)) {
                            // Do not forward the message
                            // Prevent from deadlock
                            logger.debug("Ack timeout expired");
                            this.nodeManager.noMoreNodeNeeded(nodeAvailable);
                            return;
                        }
                    }

                    if (nodeAck.intValue() != -1) {
                        // Setting vnInformation and JobId
                        if (vnName != null) {
                            try {
                                nodeAvailable.getProActiveRuntime()
                                             .registerVirtualNode(vnName, true);
                            } catch (Exception e) {
                                logger.warn("Couldn't register " + vnName +
                                    " in the PAR", e);
                            }
                        }
                        if (jobId != null) {
                            nodeAvailable.getNodeInformation().setJobID(jobId);
                        }
                        numberOfNodes = (numberOfNodes == MAX_NODE) ? MAX_NODE
                                                                    : nodeAck.intValue();
                        logger.info("Giving 1 node to vn: " + vnName);
                    } else {
                        // It's a NACK node
                        logger.debug("NACK node received");
                        this.nodeManager.noMoreNodeNeeded(nodeAvailable);
                        // No more nodes needed
                        return;
                    }
                }

                // Do we need more nodes?
                if (numberOfNodes == 0) {
                    logger.debug("No more nodes are needed");
                    return;
                }
            }
        }

        // My friend needs more nodes, so I'm broadcasting his request to my own
        // friends
        // Forwarding the message
        if (broadcast) {
            if (uuid == null) {
                logger.debug("Generating uuid for askingNode message");
                uuid = UUID.randomUUID();
            }
            this.acquaintances.askingNode(ttl, uuid, remoteService,
                numberOfNodes, lookup, vnName, jobId, nodeFamilyRegexp);
            logger.debug("Broadcast askingNode message with #" + uuid);
        }
    }

    public P2PNodeLookup getNodes(int numberOfNodes, String nodeFamilyRegexp,
        String vnName, String jobId) {
        assert vnName != null : vnName;
        assert jobId != null : jobId;
        Object[] params = new Object[5];
        params[0] = new Integer(numberOfNodes);
        params[1] = this.stubOnThis;
        params[2] = vnName;
        params[3] = jobId;
        params[4] = nodeFamilyRegexp;

        P2PNodeLookup lookup = null;
        try {
            lookup = (P2PNodeLookup) ProActive.newActive(P2PNodeLookup.class.getName(),
                    params, this.p2pServiceNode);
            ProActive.enableAC(lookup);
            if (numberOfNodes == MAX_NODE) {
                this.waitingMaximunNodesLookup.add(lookup);
            } else {
                this.waitingNodesLookup.add(lookup);
            }
        } catch (ActiveObjectCreationException e) {
            logger.fatal("Couldn't create an active lookup", e);
            return null;
        } catch (NodeException e) {
            logger.fatal("Couldn't connect node to creat", e);
            return null;
        } catch (IOException e) {
            if (logger.isDebugEnabled()) {
                logger.debug("Couldn't enable AC for a nodes lookup", e);
            }
        }

        if (logger.isInfoEnabled()) {
            if (numberOfNodes != MAX_NODE) {
                logger.info("Asking for " + numberOfNodes + " nodes");
            } else {
                logger.info("Asking for maxinum nodes");
            }
        }
        return lookup;
    }

    public P2PNodeLookup getNodes(int numberOfNodes, String vnName, String jobId) {
        return this.getNodes(numberOfNodes, ".*", vnName, jobId);
    }

    public Node getANode(String vnName, String jobId) {
        return this.stubOnThis.getANode(vnName, jobId, this.stubOnThis);
    }

    public Node getANode(String vnName, String jobId, P2PService service) {
        if (service.equals(this.stubOnThis)) {
            return this.acquaintanceManager.randomPeer()
                                           .getANode(vnName, jobId, service);
        }
        P2PNode askedNode = this.nodeManager.askingNode(null);
        Node nodeAvailable = askedNode.getNode();

        if (nodeAvailable != null) {
            if (vnName != null) {
                try {
                    nodeAvailable.getProActiveRuntime()
                                 .registerVirtualNode(vnName, true);
                } catch (Exception e) {
                    logger.warn("Couldn't register " + vnName + " in the PAR", e);
                }
            }
            if (jobId != null) {
                nodeAvailable.getNodeInformation().setJobID(jobId);
            }
            return nodeAvailable;
        }
        return this.acquaintanceManager.randomPeer()
                                       .getANode(vnName, jobId, service);
    }

    public P2PNodeLookup getMaximunNodes(String vnName, String jobId) {
        return this.getNodes(P2PConstants.MAX_NODE, vnName, jobId);
    }

    public StringWrapper getAddress() {
        return new StringWrapper(this.p2pServiceNode.getNodeInformation()
                                                    .getURL());
    }

---