org/objectweb/proactive/core/descriptor/data/VirtualNodeImpl.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.core.descriptor.data;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.rmi.AlreadyBoundException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;
import org.objectweb.proactive.ProActive;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.descriptor.services.FaultToleranceService;
import org.objectweb.proactive.core.descriptor.services.P2PDescriptorService;
import org.objectweb.proactive.core.descriptor.services.SchedulerLookupService;
import org.objectweb.proactive.core.descriptor.services.ServiceThread;
import org.objectweb.proactive.core.descriptor.services.ServiceUser;
import org.objectweb.proactive.core.descriptor.services.TechnicalService;
import org.objectweb.proactive.core.descriptor.services.UniversalService;
import org.objectweb.proactive.core.event.NodeCreationEvent;
import org.objectweb.proactive.core.event.NodeCreationEventListener;
import org.objectweb.proactive.core.event.NodeCreationEventProducerImpl;
import org.objectweb.proactive.core.event.RuntimeRegistrationEvent;
import org.objectweb.proactive.core.event.RuntimeRegistrationEventListener;
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.node.NodeImpl;
import org.objectweb.proactive.core.process.AbstractExternalProcessDecorator;
import org.objectweb.proactive.core.process.AbstractSequentialListProcessDecorator;
import org.objectweb.proactive.core.process.DependentProcess;
import org.objectweb.proactive.core.process.ExternalProcess;
import org.objectweb.proactive.core.process.ExternalProcessDecorator;
import org.objectweb.proactive.core.process.JVMProcess;
import org.objectweb.proactive.core.process.UniversalProcess;
import org.objectweb.proactive.core.process.filetransfer.FileTransferDefinition;
import org.objectweb.proactive.core.process.filetransfer.FileTransferWorkShop;
import org.objectweb.proactive.core.process.filetransfer.FileTransferDefinition.FileDescription;
import org.objectweb.proactive.core.process.mpi.MPIProcess;
import org.objectweb.proactive.core.runtime.ProActiveRuntime;
import org.objectweb.proactive.core.runtime.ProActiveRuntimeImpl;
import org.objectweb.proactive.core.runtime.RuntimeFactory;
import org.objectweb.proactive.core.util.UrlBuilder;
import org.objectweb.proactive.core.util.log.Loggers;
import org.objectweb.proactive.core.util.log.ProActiveLogger;
import org.objectweb.proactive.ext.security.ProActiveSecurityManager;
import org.objectweb.proactive.filetransfer.FileTransfer;
import org.objectweb.proactive.filetransfer.FileVector;
import org.objectweb.proactive.p2p.service.node.P2PNodeLookup;
import org.objectweb.proactive.p2p.service.util.P2PConstants;
import org.objectweb.proactive.scheduler.SchedulerConstants;


public class VirtualNodeImpl extends NodeCreationEventProducerImpl
    implements VirtualNode, Serializable, RuntimeRegistrationEventListener,
        NodeCreationEventListener, ServiceUser {

    private final static Logger P2P_LOGGER = ProActiveLogger.getLogger(Loggers.P2P_VN);
    private final static Logger FILETRANSFER_LOGGER = ProActiveLogger.getLogger(Loggers.FILETRANSFER);
    private final static Logger DEPLOYMENT_FILETRANSFER_LOGGER = ProActiveLogger.getLogger(Loggers.DEPLOYMENT_FILETRANSFER);
    public static int counter = 0;

    //
    //  ----- PRIVATE MEMBERS -----------------------------------------------------------------------------------
    //

    protected transient ProActiveRuntimeImpl proActiveRuntimeImpl;

    private String name;

    private String property;

    private java.util.ArrayList<VirtualMachine> virtualMachines;

    private java.util.ArrayList<String> localVirtualMachines;

    private int lastVirtualMachineIndex;

    private ArrayList<ProActiveRuntime> createdRuntimeF;

    private java.util.ArrayList<Node> createdNodes;

    private java.util.ArrayList<FileTransferDefinition> fileTransferDeploy;

    private java.util.ArrayList<FileTransferDefinition> fileTransferRetrieve;

    private HashMap<String, FileVector> fileTransferDeployedStatus;
    private int fileBlockSize;
    private int overlapping;

    private int lastNodeIndex;

    private int nbMappedNodes;

    private int minNumberOfNodes = 0;

    private int nbCreatedNodes;

    private boolean nodeCreated = false;
    private boolean isActivated = false;

    private Hashtable<String, VirtualMachine> awaitedVirtualNodes;
    private String registrationProtocol;
    private boolean registration = false;
    private boolean waitForTimeout = false;

    //boolean used to know if the vn is mapped only with a P2P service that request MAX nodes
    // indeed the behavior is different when returning nodes
    private boolean MAX_P2P = false;

    //protected int MAX_RETRY = 70;

    protected long timeout = 70000;

    protected long globalTimeOut;
    private Object uniqueActiveObject = null;

    // Security 
    private ProActiveSecurityManager proactiveSecurityManager;
    protected String jobID = ProActive.getJobId();

    // FAULT TOLERANCE
    private FaultToleranceService ftService;
    private Vector<Node> p2pNodes = new Vector<Node>();

    // PAD infos
    private boolean mainVirtualNode;
    private String padURL;
    private Vector<P2PNodeLookup> p2pNodeslookupList = new Vector<P2PNodeLookup>();

    //REGISTRATION ATTEMPTS
    private final int REGISTRATION_ATTEMPTS = 2;

    // MPI Process
    ExternalProcess mpiProcess = null;
    private TechnicalService technicalService;
    private String descriptorURL;

    //
    //  ----- CONSTRUCTORS -----------------------------------------------------------------------------------
    //

    VirtualNodeImpl() {
    }

    VirtualNodeImpl(String name,
        ProActiveSecurityManager proactiveSecurityManager, String padURL,
        boolean isMainVN, ProActiveDescriptor descriptor) {
        // if we launch several times the same application 
        // we have to change the name of the main VNs because of
        // the register, otherwise we will monitor each time all the last
        // main VNs with the same name.
        if (isMainVN) {
            this.name = name + "_" + (counter++);
        } else {
            this.name = name;
        }

        virtualMachines = new java.util.ArrayList<VirtualMachine>(5);
        localVirtualMachines = new java.util.ArrayList<String>();
        createdNodes = new java.util.ArrayList<Node>();
        createdRuntimeF = new ArrayList<ProActiveRuntime>();
        awaitedVirtualNodes = new Hashtable<String, VirtualMachine>();
        fileTransferDeploy = new ArrayList<FileTransferDefinition>();
        fileTransferDeployedStatus = new HashMap<String, FileVector>();
        fileTransferRetrieve = new ArrayList<FileTransferDefinition>();
        proActiveRuntimeImpl = (ProActiveRuntimeImpl) ProActiveRuntimeImpl.getProActiveRuntime();
        fileBlockSize = org.objectweb.proactive.core.filetransfer.FileBlock.DEFAULT_BLOCK_SIZE;
        overlapping = org.objectweb.proactive.core.filetransfer.FileTransferService.DEFAULT_MAX_SIMULTANEOUS_BLOCKS;

        if (logger.isDebugEnabled()) {
            logger.debug("vn " + this.name + " registered on " +
                proActiveRuntimeImpl.getVMInformation().getVMID().toString());
        }

        // SECURITY
        this.proactiveSecurityManager = proactiveSecurityManager;

        // added for main infos
        this.mainVirtualNode = isMainVN;
        this.padURL = padURL;

        this.descriptorURL = descriptor.getProActiveDescriptorURL();

    }

    //
    //  ----- PUBLIC METHODS -----------------------------------------------------------------------------------
    //

    public void setProperty(String value) {
        this.property = value;
    }

    public String getProperty() {
        return property;
    }

    public long getTimeout() {
        return timeout;
    }

    public void setTimeout(long timeout, boolean waitForTimeout) {
        this.timeout = timeout;
        this.waitForTimeout = waitForTimeout;
    }

    public void setName(String s) {
        this.name = s;
    }

    public String getName() {
        return name;
    }

    public void addVirtualMachine(VirtualMachine virtualMachine) {
        if (isActivated) {
            // Do not call this method when the VN has been activated or it will open the door
            // to various race conditions.
            logger.error(
                "addVirtualMachine() cannot be called when the virtualNode has been activated");

            return;
        }

        virtualMachines.add(virtualMachine);

        if (logger.isDebugEnabled()) {
            logger.debug("mapped VirtualNode=" + name +
                " with VirtualMachine=" + virtualMachine.getName());
        }
    }

    public void addFileTransferDeploy(FileTransferDefinition ft) {
        if (ft == null) {
            return;
        }

        fileTransferDeploy.add(ft);

        if (logger.isDebugEnabled()) {
            logger.debug("mapped VirtualNode=" + name +
                " with FileTransferDeploy id=" + ft.getId());
        }
    }

    public void addFileTransferRetrieve(FileTransferDefinition ft) {
        if (ft == null) {
            return;
        }

        fileTransferRetrieve.add(ft);

        if (logger.isDebugEnabled()) {
            logger.debug("mapped VirtualNode=" + name +
                " with FileTransferRetrieve id=" + ft.getId());
        }
    }

    public VirtualMachine getVirtualMachine() {
        if (virtualMachines.isEmpty()) {
            return null;
        }

        VirtualMachine vm = (VirtualMachine) virtualMachines.get(lastVirtualMachineIndex);

        return vm;
    }

    public VirtualMachine getVirtualMachine(String name) {
        Iterator it = virtualMachines.iterator();

        while (it.hasNext()) {
            VirtualMachine vm = (VirtualMachine) it.next();

            if (vm.getName().equals(name)) {
                return vm;
            }
        }

        return null;
    }

    public void activate() {
        if (!isActivated) {
            for (Iterator<VirtualMachine> it = virtualMachines.iterator();
                    it.hasNext();) {
                VirtualMachine vm = it.next();

                if (vm.getCreatorId() == null) {
                    vm.setCreatorId(this.name);
                } else {
                    awaitedVirtualNodes.put(vm.getCreatorId(), vm);
                }
            }

            proActiveRuntimeImpl.addRuntimeRegistrationEventListener(this);
            proActiveRuntimeImpl.registerLocalVirtualNode(this, this.name);

            for (int i = 0; i < virtualMachines.size(); i++) {
                VirtualMachine vm = getVirtualMachine();

                // first check if it is a process that is attached to the vm
                if (vm.hasProcess()) {
                    boolean vmAlreadyAssigned = !((vm.getCreatorId()).equals(this.name));

                    //boolean vmAlreadyAssigned = vm.isActivated();
                    ExternalProcess process = getProcess(vm, vmAlreadyAssigned);

                    /*   //check if it's a gLiteProcess. If it's a gLiteProcess, get the cpu number and run the glite submission command "cpuNumber" times.
                       int cpuNumber = checkGLiteProcess(process);
                       if (cpuNumber > -1) {
                           while (cpuNumber > 0) {
                               try {
                                   setParameters(process, vm);
                                   process.startProcess();
                                   process.setStarted(false);
                               } catch (IOException e) {
                                   e.printStackTrace();
                               }
                               cpuNumber--;
                           }
                       }
                     */

                    //  get the rank of sequential process - return -1 if it does not exist
                    int rankOfSequentialProcess = checkForSequentialProcess(process);

                    // there's a sequential process in the hierarchical process
                    if (rankOfSequentialProcess > -1) {
                        ExternalProcess deepCopy = (ExternalProcess) makeDeepCopy(process);

                        // there's a process before the sequential one
                        if (rankOfSequentialProcess > 0) {
                            process = getSequentialProcessInHierarchie(process,
                                    rankOfSequentialProcess);
                        }

                        // check if the first element is a Service or a process
                        if (((AbstractSequentialListProcessDecorator) process).isFirstElementIsService()) {
                            UniversalService firstService = (UniversalService) ((AbstractSequentialListProcessDecorator) process).getFirstService();

                            //   It is a service that is mapped to the vm.
                            startService(firstService, vm);
                            globalTimeOut = System.currentTimeMillis() +
                                timeout;

                            try {
                                waitForAllNodesCreation();
                            } catch (NodeException e) {
                                // TODO Auto-generated catch block
                                e.printStackTrace();
                            }
                        } else {
                            ExternalProcess firstProcess = (ExternalProcess) ((AbstractSequentialListProcessDecorator) process).getFirstProcess();

                            // build the process with the rest of hierarchie
                            if (rankOfSequentialProcess > 0) {
                                firstProcess = buildProcessWithHierarchie((ExternalProcess) makeDeepCopy(
                                            deepCopy), firstProcess,
                                        rankOfSequentialProcess);
                            }

                            setParameters(firstProcess, vm);

                            try {
                                proActiveRuntimeImpl.createVM(firstProcess);
                                globalTimeOut = System.currentTimeMillis() +
                                    timeout;
                                waitForAllNodesCreation();
                            } catch (java.io.IOException e) {
                                e.printStackTrace();
                            } catch (NodeException e1) {
                                e1.printStackTrace();
                            }
                        }

                        try {
                            ExternalProcess nextProcess = null;

                            // loop on each process in the sequence
                            while ((nextProcess = (ExternalProcess) ((AbstractSequentialListProcessDecorator) process).getNextProcess()) != null) {
                                boolean launchProcessManually = false;

                                /* if process is a dependent process then each process
                                 * in the sequence list except the first one have to receive
                                 * an array of objects relative to their dependence.
                                 * we assume that this dependence is the generation ov nodes
                                 */
                                if (process.isDependent()) {
                                    ((DependentProcess) nextProcess).setDependencyParameters(getNodes());

                                    if (nextProcess instanceof MPIProcess) {
                                        launchProcessManually = true;
                                    }
                                }

                                // rebuild the process with the rest of hierarchie
                                if (rankOfSequentialProcess > 0) {
                                    nextProcess = this.buildProcessWithHierarchie((ExternalProcess) makeDeepCopy(
                                                deepCopy), nextProcess,
                                            rankOfSequentialProcess);
                                }

                                if (!launchProcessManually) {
                                    setParameters(nextProcess, vm);
                                    proActiveRuntimeImpl.createVM(nextProcess);
                                    // initialization of the global timeout
                                    globalTimeOut = System.currentTimeMillis() +
                                        timeout;
                                    waitForAllNodesCreation();
                                } else {
                                    mpiProcess = nextProcess;
                                }
                            }
                        } catch (java.io.IOException e) {
                            e.printStackTrace();
                        } catch (NodeException e1) {
                            e1.printStackTrace();
                        }
                    } else {
                        // Test if that is this virtual Node that originates the creation of the vm
                        // else the vm was already created by another virtualNode, in that case, nothing is
                        // done at this point, nodes creation will occur when the runtime associated with the jvm
                        // will register.
                        if (!vmAlreadyAssigned) {
                            setParameters(process, vm);

                            try {
                                // It is this virtual Node that originates the creation of the vm
                                proActiveRuntimeImpl.createVM(process);
                            } catch (java.io.IOException e) {
                                e.printStackTrace();
                                logger.error("cannot activate virtualNode " +
                                    this.name + " with the process " +
                                    process.getCommand());
                            }
                        }
                    }
                } else {
                    // It is a service that is mapped to the vm.
                    startService(vm);
                }

                increaseIndex();
            }

            // local nodes creation
            for (int i = 0; i < localVirtualMachines.size(); i++) {
                String protocol = localVirtualMachines.get(i);
                internalCreateNodeOnCurrentJvm(protocol);
            }

            //initialization of the global timeout
            globalTimeOut = System.currentTimeMillis() + timeout;
            isActivated = true;

            if (registration) {
                register();
            }

            // FAULT TOLERANCE
            try {
                if (this.ftService != null) {
                    // register nodes only if ressource is not null
                    this.ftService.registerRessources(this.getNodes());
                }
            } catch (NodeException e) {
                logger.error(e.getMessage());
            }
        } else {
            logger.debug("VirtualNode " + this.name + " already activated");
        }
    }

    public ExternalProcess getMPIProcess() {
        return (ExternalProcess) makeDeepCopy(mpiProcess);
    }

    public boolean hasMPIProcess() {
        return !(mpiProcess == null);
    }

    //returns the sequential process in the process hierarchie
    private ExternalProcess getSequentialProcessInHierarchie(
        ExternalProcess process, int rank) {
        while (rank > 0) {
            process = ((AbstractExternalProcessDecorator) process).getTargetProcess();
            rank--;
        }

        return process;
    }

    // returns a process such that the target of p is finalProcess
    private ExternalProcess buildProcessWithHierarchie(
        ExternalProcess process, ExternalProcess finalProcess, int rank) {
        if (rank == 0) {
            return finalProcess;
        } else {
            ((AbstractExternalProcessDecorator) process).setTargetProcess(buildProcessWithHierarchie(
                    ((AbstractExternalProcessDecorator) process).getTargetProcess(),
                    finalProcess, rank - 1));

            return process;
        }
    }

    // returns the rank of sequential process in the processes hierarchie, -1 otherwise
    private int checkForSequentialProcess(ExternalProcess process) {
        int res = 0;

        while (!(process instanceof JVMProcess)) {
            // a sequential process was found return its rank
            if (process.isSequential()) {
                return res;
            } else {
                res++;
                process = ((ExternalProcess) ((ExternalProcessDecorator) process).getTargetProcess());
            }
        }

        return -1;
    }

    public boolean isMainVirtualNode() {
        return mainVirtualNode;
    }

    public String getPadURL() {
        return padURL;
    }

    public void setIsMainVirtualNode(boolean isMainVirtualNode) {
        mainVirtualNode = isMainVirtualNode;
    }

    public void setPadURL(String XMLpadURL) {
        this.padURL = XMLpadURL;
    }

    /*
     *
     * @see org.objectweb.proactive.core.descriptor.data.VirtualNode#getNbMappedNodes()
     */
    public int getNbMappedNodes() {
        if (isActivated) {
            return nbMappedNodes;
        } else {
            int nbMappedNodesTmp = 0;

            for (int i = 0; i < virtualMachines.size(); i++) {
                VirtualMachine vm = getVirtualMachine();

                // first check if it is a process that is attached to the vm
                if (vm.hasProcess()) {
                    ExternalProcess process = vm.getProcess();
                    int nbNodesPerCreatedVM = new Integer(vm.getNbNodesOnCreatedVMs()).intValue();

                    if (process.getNodeNumber() == UniversalProcess.UNKNOWN_NODE_NUMBER) {
                        return UniversalProcess.UNKNOWN_NODE_NUMBER;
                    } else {
                        nbMappedNodesTmp = nbMappedNodesTmp +
                            (process.getNodeNumber() * nbNodesPerCreatedVM);
                    }
                }
            }

            return nbMappedNodesTmp;
        }
    }

    public int createdNodeCount() {
        throw new RuntimeException(
            "This method is deprecated, use getNumberOfCurrentlyCreatedNodes() or getNumberOfCreatedNodesAfterDeployment()");
    }

    /*
     *  (non-Javadoc)
     * @see org.objectweb.proactive.core.descriptor.data.VirtualNode#getNumberOfCurrentlyCreatedNodes()
     */
    public int getNumberOfCurrentlyCreatedNodes() {
        return nbCreatedNodes;
    }

    /*
     *  (non-Javadoc)
     * @see org.objectweb.proactive.core.descriptor.data.VirtualNode#getNumberOfCreatedNodesAfterDeployment()
     */
    public int getNumberOfCreatedNodesAfterDeployment() {
        try {
            waitForAllNodesCreation();
        } catch (NodeException e) {
            logger.error("Problem occured while waiting for nodes creation");
        }

        return nbCreatedNodes;
    }

    public Node getNode() throws NodeException {
        //try first to get the Node from the createdNodes array to be continued
        Node node;
        waitForNodeCreation();

        if (!createdNodes.isEmpty()) {
            node = createdNodes.get(lastNodeIndex);
            increaseNodeIndex();

            //wait for pending file transfer
            FileVector fw = fileTransferDeployedStatus.get(node.getNodeInformation()
                                                               .getName());

            if (fw != null) {
                fw.waitForAll(); //wait-by-necessity
            }

            return node;
        } else {
            throw new NodeException("Cannot get a node from Virtual Node " + this.name+". Descriptor in use : \""+descriptorURL+"\"." );
        }
    }

    @Deprecated
    public Node getNode(int index) throws NodeException {
        Node node = createdNodes.get(index);

        if (node == null) {
            throw new NodeException(
                "Cannot return the first node, no nodes hava been created for Virtual Node "+this.name+". Descriptor in use : \""+descriptorURL+"\".");
        }

        FileVector fw = fileTransferDeployedStatus.get(node.getNodeInformation()
                                                           .getName());

        if (fw != null) {
            fw.waitForAll(); //wait-by-necessity
        }

        return node;
    }

    public String[] getNodesURL() throws NodeException {
        String[] nodeNames;

        try {
            waitForAllNodesCreation();
        } catch (NodeException e) {
            logger.error(e.getMessage());
        }

        if (!createdNodes.isEmpty()) {
            synchronized (createdNodes) {
                nodeNames = new String[createdNodes.size()];

                for (int i = 0; i < createdNodes.size(); i++) {
                    nodeNames[i] = createdNodes.get(i).getNodeInformation()
                                               .getURL();
                }
            }
        } else {
            if (!MAX_P2P) {
                throw new NodeException(
                    "Cannot return nodes, no nodes have been created for Virtual Node "+this.name+". Descriptor in use : \""+descriptorURL+"\".");
            } else {
                logger.warn("WARN: No nodes have yet been created.");
                logger.warn(
                    "WARN: This behavior might be normal, since P2P service is used with MAX number of nodes requested");
                logger.warn("WARN: Returning empty array");

                return new String[0];
            }
        }

        return nodeNames;
    }

    public Node[] getNodes() throws NodeException {
        Node[] nodeTab;

        try {
            waitForAllNodesCreation();
        } catch (NodeException e) {
            logger.error(e.getMessage());
        }

        if (!createdNodes.isEmpty()) {
            synchronized (createdNodes) {
                nodeTab = new Node[createdNodes.size()];

                for (int i = 0; i < createdNodes.size(); i++) {
                    nodeTab[i] = createdNodes.get(i);
                }
            }
        } else {
            if (!MAX_P2P) {
                throw new NodeException(
                    "Cannot return nodes, no nodes have been created for Virtual Node "+this.name+". Descriptor in use : \""+descriptorURL+"\".");
            } else {
                logger.warn("WARN: No nodes have yet been created.");
                logger.warn(
                    "WARN: This behavior might be normal, since P2P service is used with MAX number of nodes requested");
                logger.warn("WARN: Returning empty array");

                return new Node[0];
            }
        }

        return nodeTab;
    }

    public Node getNode(String url) throws NodeException {
        Node node = null;

        try {
            waitForAllNodesCreation();
        } catch (NodeException e) {
            logger.error(e.getMessage());
        }

        if (!createdNodes.isEmpty()) {
            synchronized (createdNodes) {
                for (int i = 0; i < createdNodes.size(); i++) {
                    if (createdNodes.get(i).getNodeInformation().getURL()
                                        .equals(url)) {
                        node = createdNodes.get(i);

                        break;
                    }
                }

                return node;
            }
        } else {
            throw new NodeException(
                "Cannot return nodes, no nodes hava been created. Descriptor in use : \""+descriptorURL+"\".");
        }
    }

    public void killAll(boolean softly) {
        Node node;
        ProActiveRuntime part = null;

        if (isActivated) {
            // Killing p2p nodes
            if (this.p2pNodeslookupList.size() > 0) {
                for (int index = 0; index < this.p2pNodeslookupList.size();
                        index++) {
                    P2PNodeLookup currentNodesLookup = this.p2pNodeslookupList.get(index);
                    currentNodesLookup.killAllNodes();
                }
            }

            // Killing other nodes
            for (int i = 0; i < createdNodes.size(); i++) {
                node = createdNodes.get(i);
                part = node.getProActiveRuntime();

                if (this.p2pNodes.contains(node)) {
                    continue;
                }

                //we have to be carefull. Indeed if the node is local, we do not
                // want to kill the runtime, otherwise the application is over
                // so if the node is local, we just unregister this node from any registry
                if (!NodeFactory.isNodeLocal(node)) {
                    try {
                        part.killRT(softly);
                    } catch (ProActiveException e1) {
                        e1.printStackTrace();
                    } catch (Exception e) {
                        logger.info(" Virtual Machine " +
                            part.getVMInformation().getVMID() + " on host " +
                            UrlBuilder.getHostNameorIP(
                                part.getVMInformation().getInetAddress()) +
                            " terminated!!!");
                    }
                } else {
                    try {
                        //                                      the node is local, unregister it.
                        part.killNode(node.getNodeInformation().getURL());
                    } catch (ProActiveException e) {
                        e.printStackTrace();
                    }
                }
            }

            isActivated = false;

            try {
                //if registered in any regigistry, unregister everything
                if (registration) {
                    ProActive.unregisterVirtualNode(this);
                }
                //else unregister just in the local runtime
                else {
                    proActiveRuntimeImpl.unregisterVirtualNode(this.name);
                }
            } catch (ProActiveException e) {
                e.printStackTrace();
            }

            // if not activated unregister it from the local runtime
        } else {
            proActiveRuntimeImpl.unregisterVirtualNode(this.name);
        }

        for (int i = 0; i < createdRuntimeF.size(); i++) {
            part = createdRuntimeF.get(i);

            try {
                part.killRT(true);
            } catch (Exception e) {
                logger.info(" Forwarder " + part.getVMInformation().getVMID() +
                    " on host " +
                    UrlBuilder.getHostNameorIP(part.getVMInformation()
                                                   .getInetAddress()) +
                    " terminated!!!");
            }
        }
    }

    public void createNodeOnCurrentJvm(String protocol) {
        if (protocol == null) {
            protocol = System.getProperty("proactive.communication.protocol");
        }

        localVirtualMachines.add(protocol);
    }

    public Object getUniqueAO() throws ProActiveException {
        if (!property.equals("unique_singleAO")) {
            logger.warn(
                "!!!!!!!!!!WARNING. This VirtualNode is not defined with unique_single_AO property in the XML descriptor. Calling getUniqueAO() on this VirtualNode can lead to unexpected behaviour");
        }

        if (uniqueActiveObject == null) {
            try {
                Node node = getNode();

                if (node.getActiveObjects().length > 1) {
                    logger.warn(
                        "!!!!!!!!!!WARNING. More than one active object is created on this VirtualNode.");
                }

                uniqueActiveObject = node.getActiveObjects()[0];
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        if (uniqueActiveObject == null) {
            throw new ProActiveException(
                "No active object are registered on this VirtualNode");
        }

        return uniqueActiveObject;
    }

    public boolean isActivated() {
        return isActivated;
    }

    public boolean isLookup() {
        return false;
    }

    //
    //-------------------IMPLEMENTS Job-----------------------------------
    //

    public String getJobID() {
        return this.jobID;
    }
    
    //
    //-------------------IMPLEMENTS RuntimeRegistrationEventListener------------
    //
    private transient ExecutorService rrThreadpool = Executors.newCachedThreadPool();


    //
    //-------------------IMPLEMENTS RuntimeRegistrationEventListener------------
    //
    public void runtimeRegistered(RuntimeRegistrationEvent event) {
        if (event.getType() == RuntimeRegistrationEvent.FORWARDER_RUNTIME_REGISTERED) {
            forwarderRuntimeRegisteredPerform(event);
        } else {
            // Try to avoid the contention at EventListener level
            // A listener shouldn't perform long computation, so a thread is spawned
            rrThreadpool.execute(new RuntimeRegisteredPerform(event));
        }
    }

    private class RuntimeRegisteredPerform implements Runnable {
        private RuntimeRegistrationEvent event;

        public RuntimeRegisteredPerform(RuntimeRegistrationEvent _event) {
            this.event = _event;
        }

        public void run() {
            runtimeRegisteredPerform(event);
        }
    }
    
    private synchronized void forwarderRuntimeRegisteredPerform(
        RuntimeRegistrationEvent event) {
        VirtualMachine virtualMachine = null;

        for (int i = 0; i < virtualMachines.size(); i++) {
            if (((VirtualMachine) virtualMachines.get(i)).getName()
                     .equals(event.getVmName())) {
                virtualMachine = (VirtualMachine) virtualMachines.get(i);
            }
        }

        //Check if it this virtualNode that originates the process
        if ((event.getCreatorID().equals(this.name)) &&
                (virtualMachine != null)) {
            if (logger.isDebugEnabled()) {
                logger.debug("forwarder " + event.getCreatorID() +
                    " registered on virtualnode " + this.name);
            }
        }

        createdRuntimeF.add(event.getRegisteredRuntime());
    }

    private void runtimeRegisteredPerform(RuntimeRegistrationEvent event) {
        VirtualMachine virtualMachine = null;

        // Not synchronized because it is now impossible to add a new VirtualMachine
        for (int i = 0; i < virtualMachines.size(); i++) {
            if (((VirtualMachine) virtualMachines.get(i)).getName()
                     .equals(event.getVmName())) {
                virtualMachine = (VirtualMachine) virtualMachines.get(i);
            }
        }

        //Check if it this virtualNode that originates the process
        if ((event.getCreatorID().equals(this.name)) &&
                (virtualMachine != null)) {
            if (logger.isDebugEnabled()) {
                logger.debug("runtime " + event.getCreatorID() +
                    " registered on virtualnode " + this.name);
            }

            createNodes(event, virtualMachine);
        }

        // Check if the virtualNode that originates the process is among awaited
        // VirtualNodes
        if (awaitedVirtualNodes.containsKey(event.getCreatorID())) {
            createNodes(event, awaitedVirtualNodes.get(event.getCreatorID()));
        }
    }

    private void createNodes(RuntimeRegistrationEvent event,
        VirtualMachine _virtualMachine) {
        String protocol;
        ProActiveRuntime proActiveRuntimeRegistered;
        String nodeHost;
        int port;
        String url;

        protocol = event.getProtocol();
        proActiveRuntimeRegistered = event.getRegisteredRuntime();
        nodeHost = proActiveRuntimeRegistered.getVMInformation().getHostName();
        port = UrlBuilder.getPortFromUrl(proActiveRuntimeRegistered.getURL());
        url = null;

        try {
            int nodeNumber = (new Integer((String) _virtualMachine.getNbNodesOnCreatedVMs())).intValue();

            for (int i = 1; i <= nodeNumber; i++) {
                ProActiveSecurityManager siblingPSM = null;

                if (proactiveSecurityManager != null) {
                    siblingPSM = proactiveSecurityManager.generateSiblingCertificate(this.name);
                }

                int registerAttempts = REGISTRATION_ATTEMPTS;

                while (registerAttempts > 0) { // If there is an
                                               // AlreadyBoundException, we
                                               // will gerate an other
                                               // random node's name and
                                               // try to register it again
                    String nodeName = this.name +
                        Integer.toString(ProActiveRuntimeImpl.getNextInt());
                    url = buildURL(nodeHost, nodeName, protocol, port);

                    // nodes are created from the registered runtime, since
                    // this
                    // virtualNode is
                    // waiting for runtime registration to perform
                    // co-allocation
                    // in the jvm.
                    try {
                        proActiveRuntimeRegistered.createLocalNode(url, false,
                            siblingPSM, this.getName(), this.jobID);

                        // Registration has succeded.
                        performOperations(proActiveRuntimeRegistered, url,
                            protocol, event.getVmName());

                        break;
                    } catch (AlreadyBoundException e) {
                        registerAttempts--;
                    }
                }

                if (registerAttempts == 0) {
                    logger.warn("createLocalNode failed " +
                        REGISTRATION_ATTEMPTS + " times for runtime " +
                        nodeHost);
                }
            }
        } catch (ProActiveException e) {
            e.printStackTrace();
        }
    }

    public void setRuntimeInformations(String information, String value)
        throws ProActiveException {
        //        try {
        //            checkProperty(information);
        //        } catch (ProActiveException e) {
        //            throw new ProActiveException("No property can be set at runtime on this VirtualNode",
        //                e);
        //        }
        //No need to check if the property exist since no property can be set 
        // at runtime on a VNImpl. This might change in the future.
        throw new ProActiveException(
            "No property can be set at runtime on this VirtualNode");
    }

    public void setService(UniversalService service) throws ProActiveException {
        if (FaultToleranceService.FT_SERVICE_NAME.equals(
                    service.getServiceName())) {
            this.ftService = (FaultToleranceService) service;
        } else {
            throw new ProActiveException(
                " Unable to bind the given service to a virtual node");
        }
    }

    public String getUserClass() {
        return this.getClass().getName();
    }

    public void setRegistrationProtocol(String protocol) {
        setRegistrationValue(true);
        this.registrationProtocol = protocol;
    }

    public String getRegistrationProtocol() {
        return this.registrationProtocol;
    }

    public void setMinNumberOfNodes(int min) {
        this.minNumberOfNodes = min;
    }

    public int getMinNumberOfNodes() {
        return minNumberOfNodes;
    }

    public boolean isMultiple() {
        return ((virtualMachines.size() + localVirtualMachines.size()) > 1);
    }

    //
    //-------------------PRIVATE METHODS--------------------------------------
    //
    private void internalCreateNodeOnCurrentJvm(String protocol) {
        try {
            // this method should be called when in the xml document the tag currenJVM is encountered. It means that one node must be created
            // on the jvm that originates the creation of this virtualNode(the current jvm) and mapped on this virtualNode
            // we must increase the node count
            String url = null;
            increaseNumberOfNodes(1);

            // get the Runtime for the given protocol
            ProActiveRuntime defaultRuntime = RuntimeFactory.getProtocolSpecificRuntime(checkProtocol(
                        protocol));

            //create the node
            ProActiveSecurityManager siblingPSM = null;

            if (proactiveSecurityManager != null) {
                siblingPSM = proactiveSecurityManager.generateSiblingCertificate(this.name);
            }

            int registrationAttempts = REGISTRATION_ATTEMPTS;

            while (registrationAttempts > 0) { //If there is an AlreadyBoundException, we generate an other random node name
                String nodeName = this.name +
                    Integer.toString(ProActiveRuntimeImpl.getNextInt());

                try {
                    url = defaultRuntime.createLocalNode(nodeName, false,
                            siblingPSM, this.getName(), ProActive.getJobId());
                    registrationAttempts = 0;
                } catch (AlreadyBoundException e) {
                    registrationAttempts--;
                }
            }

            //add this node to this virtualNode
            performOperations(defaultRuntime, url, protocol,
                defaultRuntime.getVMInformation().getDescriptorVMName());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private synchronized void waitForNodeCreation() throws NodeException {
        while (!nodeCreated) {
            if (!timeoutExpired()) {
                try {
                    wait(getTimeToSleep());
                } catch (InterruptedException e2) {
                    e2.printStackTrace();
                } catch (IllegalStateException e) {
                    // it may happen that we entered in the loop and just after
                    // the timeToSleep is < 0. It means that the timeout expired
                    // that is why we catch the runtime exception
                    throw new NodeException(
                        "After many retries, not even one node can be found for Virtual Node \"" +
                        this.name + "\" in descriptor \"" + this.descriptorURL +
                        "\".");
                }
            } else {
                throw new NodeException(
                    "After many retries, not even one node can be found for Virtual Node \"" +
                    this.name + "\" in descriptor \"" + this.descriptorURL + "\".");
            }
        }

        return;
    }

    public void waitForAllNodesCreation() throws NodeException {
        int tempNodeCount = nbMappedNodes;

        if (tempNodeCount != P2PConstants.MAX_NODE) {
            //nodeCount equal 0 means there is only a P2P service with MAX number of nodes requested
            // so if different of 0, we can set to false the boolean
            MAX_P2P = false;
        }

        if (waitForTimeout) {
            try {
                Thread.sleep(timeout);
            } catch (InterruptedException e2) {
                e2.printStackTrace();
            }
        } else {
            //behavior has to be moved in a synchronized method to avoid useless lock
            // when sleeping
            internalWait(tempNodeCount);
        }

        //wait for the nodes to complete their deployment file transfer
        Collection<FileVector> c = fileTransferDeployedStatus.values();
        Iterator<FileVector> it = c.iterator();

        while (it.hasNext()) {
            FileVector fw = it.next();
            fw.waitForAll(); //wait-by-necessity
        }

//        rrThreadpool.shutdown();
//        while (!rrThreadpool.isTerminated()) {
//              try {
//                Thread.sleep(100);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
        return;
    }

    private synchronized void internalWait(int tempNodeCount)
        throws NodeException {
        // check if we can release the vn before all nodes expected, are created
        // i.e the minNumber of nodes is set
        if (minNumberOfNodes != 0) {
            tempNodeCount = minNumberOfNodes;
        }

        while (nbCreatedNodes < tempNodeCount) {
            if (!timeoutExpired()) {
                try {
                    wait(getTimeToSleep());
                } catch (InterruptedException e2) {
                    e2.printStackTrace();
                } catch (IllegalStateException e) {
                    // it may happen that we entered in the loop and just after
                    // the timeToSleep is < 0. It means that the timeout expired
                    // that is why we catch the runtime exception
                    throw new NodeException("After many retries, only " +
                        nbCreatedNodes + " nodes are created on " +
                        tempNodeCount + " expected for Virtual Node \"" +
                        this.name + "\" in descriptor \"" + descriptorURL + "\".");
                }
            } else {
                throw new NodeException("After many retries, only " +
                    nbCreatedNodes + " nodes are created on " + tempNodeCount +
                    " expected for Virtual Node \"" + this.name +
                    "\" in descriptor \"" + descriptorURL + "\".");
            }
        }
    }

    private boolean timeoutExpired() {
        long currentTime = System.currentTimeMillis();

        return (globalTimeOut < currentTime);
    }

    private long getTimeToSleep() {
        // if timeToSleep is < 0 we throw an exception
        long timeToSleep = globalTimeOut - System.currentTimeMillis();

        if (timeToSleep > 0) {
            return timeToSleep;
        } else {
            throw new IllegalStateException("Timeout expired");
        }
    }

    private ExternalProcess getProcess(VirtualMachine vm,
        boolean vmAlreadyAssigned) {
        ExternalProcess copyProcess;

        //VirtualMachine vm = getVirtualMachine();
        ExternalProcess process = vm.getProcess();

        // we need to do a deep copy of the process otherwise,
        //modifications will be applied on one object that might 
        // be referenced by other virtualNodes .i.e check started
        if (!vmAlreadyAssigned) {
            copyProcess = (ExternalProcess) makeDeepCopy(process);
            vm.setProcess(copyProcess);

            return copyProcess;
        } else {
            //increment the node count by askedNodes
            increaseNumberOfNodes(process.getNodeNumber() * new Integer(
                    vm.getNbNodesOnCreatedVMs()).intValue());

            return process;
        }
    }

    private void setParameters(ExternalProcess process, VirtualMachine vm) {
        JVMProcess jvmProcess;

        //jobID = ProActive.getJobId();
        String protocolId = "";
        int nodeNumber = new Integer(vm.getNbNodesOnCreatedVMs()).intValue();

        if (logger.isDebugEnabled()) {
            logger.debug("asked for " + nodeNumber + " nodes");
        }

        protocolId = process.getProcessId();

        int cnt = process.getNodeNumber();

        if (cnt == UniversalProcess.UNKNOWN_NODE_NUMBER) {
            waitForTimeout = true;
        } else {
            increaseNumberOfNodes(cnt * nodeNumber);
        }

        //When the virtualNode will be activated, it has to launch the process
        //with such parameter.See StartRuntime
        jvmProcess = (JVMProcess) process.getFinalProcess();
        jvmProcess.setJvmOptions("-Dproactive.groupInformation=" +
            ProActiveRuntimeImpl.getProActiveRuntime().getVMInformation()
                                .getVMID().toString() + "~" +
            jvmProcess.getNewGroupId());

        //if the target class is StartRuntime, then give parameters otherwise keep parameters
        if (jvmProcess.getClassname()
                          .equals("org.objectweb.proactive.core.runtime.StartRuntime")) {
            String vnName = this.name;

            String localruntimeURL = null;

            try {
                localruntimeURL = RuntimeFactory.getDefaultRuntime().getURL();

                if (process.getUsername() != null) {
                    localruntimeURL = System.getProperty("user.name") + "@" +
                        localruntimeURL;
                }
            } catch (ProActiveException e) {
                e.printStackTrace();
            }

            if (logger.isDebugEnabled()) {
                logger.debug(localruntimeURL);
            }

            // if it is a main node we set the property for retrieving the pad
            if (mainVirtualNode || process.isHierarchical()) {
                jvmProcess.setJvmOptions(" -Dproactive.pad=" + padURL);
            }

            jvmProcess.setJvmOptions("-Dproactive.jobid=" + jobID);
            jvmProcess.setParameters(vnName + " " + localruntimeURL + " " +
                nodeNumber + " " + protocolId + " " + vm.getName());

            // FAULT TOLERANCE settings
            if (this.ftService != null) {
                jvmProcess.setJvmOptions(this.ftService.buildParamsLine());
            }
        }

        /* Setting the file transfer definitions associated with the current process,
         * and defined at the process level.
         */
        FileTransferWorkShop ftsDeploy = process.getFileTransferWorkShopDeploy();
        FileTransferWorkShop ftsRetrieve = process.getFileTransferWorkShopRetrieve();

        if ((ftsDeploy != null) && ftsDeploy.isImplicit()) {
            for (int i = 0; i < fileTransferDeploy.size(); i++)
                ftsDeploy.addFileTransfer(fileTransferDeploy.get(i));
        }

        if ((ftsRetrieve != null) && ftsRetrieve.isImplicit()) {
            for (int i = 0; i < fileTransferRetrieve.size(); i++)
                ftsRetrieve.addFileTransfer(fileTransferRetrieve.get(i));
        }
    }

    private Object makeDeepCopy(Object process) {
        //deepCopyTag = true;
        Object result = null;

        try {
            java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream();
            java.io.ObjectOutputStream oos = new java.io.ObjectOutputStream(baos);
            oos.writeObject(process);
            oos.flush();
            oos.close();

            java.io.ByteArrayInputStream bais = new java.io.ByteArrayInputStream(baos.toByteArray());
            java.io.ObjectInputStream ois = new java.io.ObjectInputStream(bais);
            result = ois.readObject();
            ois.close();
        } catch (Exception e) {
            e.printStackTrace();
        }

        //deepCopyTag = false;
        return result;
    }

    private String buildURL(String host, String name, String protocol, int port) {
        if (port != 0) {
            return UrlBuilder.buildUrl(host, name, protocol, port);
        } else {
            return UrlBuilder.buildUrl(host, name, protocol);
        }
    }

    private void increaseIndex() {
        if (virtualMachines.size() > 1) {
            lastVirtualMachineIndex = (lastVirtualMachineIndex + 1) % virtualMachines.size();
        }
    }

    private void increaseNumberOfNodes(int n) {
        nbMappedNodes = nbMappedNodes + n;

        if (logger.isDebugEnabled()) {
            logger.debug("Number of nodes = " + nbMappedNodes);
        }
    }

    private void increaseNodeIndex() {
        if (createdNodes.size() > 1) {
            lastNodeIndex = (lastNodeIndex + 1) % createdNodes.size();
        }
    }

    private String checkProtocol(String protocol) {
        if (protocol.indexOf(":") == -1) {
            return protocol.concat(":");
        }

        return protocol;
    }

    private synchronized void performOperations(ProActiveRuntime part,
        String url, String protocol, String vmName) {
        Node node = new NodeImpl(part, url, checkProtocol(protocol),
                this.jobID, vmName);
       synchronized (createdNodes) {
           createdNodes.add(node);      
        }
       
        logger.info("**** Mapping VirtualNode " + this.name + " with Node: " +
            url + " done");
        nodeCreated = true;
        nbCreatedNodes++;

        //Performe FileTransferDeploy (if needed)       
        FileVector fw = fileTransferDeploy(node);
        this.fileTransferDeployedStatus.put(node.getNodeInformation().getName(),
            fw);

        if (this.technicalService != null) {
            this.technicalService.apply(node);
        }

        // wakes up Thread that are waiting for the node creation 
        notifyAll();

        //notify all listeners that a node has been created
        notifyListeners(this, NodeCreationEvent.NODE_CREATED, node,
            nbCreatedNodes);
    }

    private void register() {
        try {
            waitForAllNodesCreation();

            //          ProActiveRuntime part = RuntimeFactory.getProtocolSpecificRuntime(registrationProtocol);
            //          part.registerVirtualnode(this.name,false);
            ProActive.registerVirtualNode(this, registrationProtocol, false);
        } catch (NodeException e) {
            logger.error(e.getMessage());
        } catch (ProActiveException e) {
            e.printStackTrace();
        } catch (AlreadyBoundException e) {
            logger.warn("The Virtual Node name " + this.getName() +
                " is already bound in the registry", e);
        }
    }

    private void setRegistrationValue(boolean value) {
        this.registration = value;
    }

    private void startService(VirtualMachine vm) {
        UniversalService service = vm.getService();

        //we need to perform a deep copy. Indeed if several vm reference
        // the same service this might lead to unexpected behaviour
        UniversalService copyService = (UniversalService) makeDeepCopy(service);
        vm.setService(copyService);

        if (service.getServiceName().equals(P2PConstants.P2P_NODE_NAME)) {
            int nodeRequested = ((P2PDescriptorService) service).getNodeNumber();

            // if it is a P2Pservice we must increase the node count with the number
            // of nodes requested
            if (nodeRequested != ((P2PDescriptorService) service).getMAX()) {
                increaseNumberOfNodes(nodeRequested);

                //nodeRequested = MAX means that the service will try to get every nodes 
                // it can. So we can't predict how many nodes will return.
            } else {
                MAX_P2P = true;
            }
        } else if (service.getServiceName()
                              .equals(SchedulerConstants.SCHEDULER_NODE_NAME)) {
            int nodeRequested = ((SchedulerLookupService) service).getNodeNumber();
            increaseNumberOfNodes(nodeRequested);
        } else {
            //increase with 1 node
            increaseNumberOfNodes(1);
        }

        new ServiceThread(this, vm).start();
    }

    private void startService(UniversalService service, VirtualMachine vm) {
        //we need to perform a deep copy. Indeed if several vm reference
        // the same service this might lead to unexpected behaviour
        UniversalService copyService = (UniversalService) makeDeepCopy(service);
        vm.setService(copyService);

        if (service.getServiceName().equals(P2PConstants.P2P_NODE_NAME)) {
            int nodeRequested = ((P2PDescriptorService) service).getNodeNumber();

            // if it is a P2Pservice we must increase the node count with the number
            // of nodes requested
            if (nodeRequested != ((P2PDescriptorService) service).getMAX()) {
                increaseNumberOfNodes(nodeRequested);

                //nodeRequested = MAX means that the service will try to get every nodes 
                // it can. So we can't predict how many nodes will return.
            } else {
                MAX_P2P = true;
            }
        } else if (service.getServiceName()
                              .equals(SchedulerConstants.SCHEDULER_NODE_NAME)) {
            int nodeRequested = ((SchedulerLookupService) service).getNodeNumber();
            increaseNumberOfNodes(nodeRequested);
        } else {
            //increase with 1 node
            increaseNumberOfNodes(1);
        }

        new ServiceThread(this, vm).start();
    }

    private void writeObject(java.io.ObjectOutputStream out)
        throws java.io.IOException {
        if (isActivated) {
            try {
                waitForAllNodesCreation();
            } catch (NodeException e) {
                out.defaultWriteObject();

                return;
            }
        }
        
        out.defaultWriteObject();
    }

    private void readObject(java.io.ObjectInputStream in)
        throws java.io.IOException, ClassNotFoundException {
        in.defaultReadObject();
        this.proActiveRuntimeImpl = (ProActiveRuntimeImpl) ProActiveRuntimeImpl.getProActiveRuntime();
        rrThreadpool = Executors.newCachedThreadPool();
    }

    // -------------------------------------------------------------------------
    // For P2P nodes acquisition
    // -------------------------------------------------------------------------

    public synchronized void nodeCreated(NodeCreationEvent event) {
        Node newNode = event.getNode();
        this.createdNodes.add(newNode);
        this.p2pNodes.add(newNode);
        nbCreatedNodes++;
        nodeCreated = true;

        if (this.technicalService != null) {
            this.technicalService.apply(newNode);
        }

        //notify all listeners that a node has been created
        notifyAll();
        notifyListeners(this, NodeCreationEvent.NODE_CREATED, newNode,
            nbCreatedNodes);
    }

    public void addP2PNodesLookup(P2PNodeLookup nodesLookup) {
        this.p2pNodeslookupList.add(nodesLookup);
        P2P_LOGGER.debug("A P2P nodes lookup added to the vn: " + this.name);
    }

    public FileVector fileTransferRetrieve()
        throws IOException, ProActiveException {
        Node[] nodes;
        FileVector fileVector = new FileVector();

        try {
            nodes = getNodes();
        } catch (NodeException e) {
            throw new ProActiveException(
                "Can not Retrieve Files, since no nodes where created for Virtual Node" +
                this.getName());
        }

        if (FILETRANSFER_LOGGER.isDebugEnabled()) {
            FILETRANSFER_LOGGER.debug("Retrieving files for " + nodes.length +
                " node(s).");
        }

        /* For all the nodes we get the VirtualMachine that spawned it, and
         * then the process linked with this VirtualMachine. We then obtain
         * the FileTransfer Retrieve Workshop from the process and using
         * the FileTransfer API we retrieve the Files.
         */
        for (int i = 0; i < nodes.length; i++) {
            String vmName = nodes[i].getNodeInformation().getDescriptorVMName();
            VirtualMachine vm = getVirtualMachine(vmName);

            if (vm == null) {
                if (FILETRANSFER_LOGGER.isDebugEnabled()) {
                    FILETRANSFER_LOGGER.info("No VM found with name: " +
                        vmName + " for node: " +
                        nodes[i].getNodeInformation().getName());
                }

                continue;
            }

            //TODO We only get the VN for the first process in the chain. We should check if it is a SSH RSH,etc...
            ExternalProcess eProcess = vm.getProcess();

            if (eProcess == null) {
                if (FILETRANSFER_LOGGER.isDebugEnabled()) {
                    FILETRANSFER_LOGGER.debug("No Process linked with VM: " +
                        vmName + " for node: " +
                        nodes[i].getNodeInformation().getName());
                }

                continue;
            }

            FileTransferWorkShop ftwRetrieve = eProcess.getFileTransferWorkShopRetrieve();
            FileDescription[] fd = ftwRetrieve.getAllFileDescriptions();

            File[] srcFile = new File[fd.length];
            File[] dstFile = new File[fd.length];

            for (int j = 0; j < fd.length; j++) {
                srcFile[j] = new File(ftwRetrieve.getAbsoluteSrcPath(fd[j]));
                dstFile[j] = new File(ftwRetrieve.getAbsoluteDstPath(fd[j]) +
                        "-" + nodes[i].getNodeInformation().getName());
            }

            long init = System.currentTimeMillis();
            fileVector.add(FileTransfer.pullFiles(nodes[i], srcFile, dstFile,
                    fileBlockSize, overlapping));

            if (FILETRANSFER_LOGGER.isDebugEnabled()) {
                FILETRANSFER_LOGGER.debug("Returned pullFiles in:" +
                    (System.currentTimeMillis() - init));
            }
        }

        return fileVector;
    }

    private FileVector fileTransferDeploy(Node node) {
        FileVector fileWrapper = new FileVector();

        if (DEPLOYMENT_FILETRANSFER_LOGGER.isDebugEnabled()) {
            DEPLOYMENT_FILETRANSFER_LOGGER.debug(
                "File Transfer Deploy files for node" +
                node.getNodeInformation().getName());
        }

        String vmName = node.getNodeInformation().getDescriptorVMName();
        VirtualMachine vm = getVirtualMachine(vmName);

        if (vm == null) {
            if (DEPLOYMENT_FILETRANSFER_LOGGER.isDebugEnabled()) {
                DEPLOYMENT_FILETRANSFER_LOGGER.debug("No VM found with name: " +
                    vmName + " for node: " +
                    node.getNodeInformation().getName());
            }

            return fileWrapper;
        }

        //TODO We only get the VN for the first process in the chain. We should check if it is a SSH, SSH, etc...
        ExternalProcess eProcess = vm.getProcess();

        if (eProcess == null) {
            if (DEPLOYMENT_FILETRANSFER_LOGGER.isDebugEnabled()) {
                DEPLOYMENT_FILETRANSFER_LOGGER.debug(
                    "No Process linked with VM: " + vmName + " for node: " +
                    node.getNodeInformation().getName());
            }

            return fileWrapper;
        }

        //if the process handled the FileTransfer we have nothing to do
        if (!eProcess.isRequiredFileTransferDeployOnNodeCreation()) {
            if (DEPLOYMENT_FILETRANSFER_LOGGER.isDebugEnabled()) {
                DEPLOYMENT_FILETRANSFER_LOGGER.debug(
                    "No ProActive FileTransfer API is required for this node.");
            }

            return fileWrapper;
        }

        FileTransferWorkShop ftwDeploy = eProcess.getFileTransferWorkShopDeploy();
        FileDescription[] fd = ftwDeploy.getAllFileDescriptions();

        if (DEPLOYMENT_FILETRANSFER_LOGGER.isDebugEnabled()) {
            DEPLOYMENT_FILETRANSFER_LOGGER.debug("Transfering " + fd.length +
                " file(s)");
        }

        File[] filesSrc = new File[fd.length];
        File[] filesDst = new File[fd.length];

        for (int j = 0; j < fd.length; j++) {
            File srcFile = new File(ftwDeploy.getAbsoluteSrcPath(fd[j]));
            File dstFile = new File(ftwDeploy.getAbsoluteDstPath(fd[j]));
            filesSrc[j] = srcFile;
            filesDst[j] = dstFile;
        }

        try {
            fileWrapper.add(FileTransfer.pushFiles(node, filesSrc, filesDst,
                    fileBlockSize, overlapping));
        } catch (Exception e) {
            logger.error("Unable to pushFile files to node " +
                node.getNodeInformation().getName());

            if (DEPLOYMENT_FILETRANSFER_LOGGER.isDebugEnabled()) {
                DEPLOYMENT_FILETRANSFER_LOGGER.debug(e.getStackTrace());
            }
        }

        return fileWrapper;
    }

    public void setFileTransferParams(int fileBlockSize, int overlapping) {
        this.fileBlockSize = fileBlockSize;
        this.overlapping = overlapping;
    }

    public void addTechnicalService(TechnicalService technicalWrapper) {
        this.technicalService = technicalWrapper;
    }

    public ArrayList getVirtualMachines() {
        return this.virtualMachines;
    }

}

---