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

import java.util.Vector;

import org.apache.log4j.Logger;
import org.objectweb.proactive.ProActive;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.descriptor.data.VirtualMachine;
import org.objectweb.proactive.core.descriptor.data.VirtualNode;
import org.objectweb.proactive.core.descriptor.data.VirtualNodeImpl;
import org.objectweb.proactive.core.event.NodeCreationEvent;
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.runtime.ProActiveRuntime;
import org.objectweb.proactive.core.runtime.ProActiveRuntimeImpl;
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.p2p.service.P2PService;
import org.objectweb.proactive.p2p.service.node.P2PNodeLookup;
import org.objectweb.proactive.p2p.service.util.P2PConstants;
import org.objectweb.proactive.scheduler.SchedulerConstants;


public class ServiceThread extends Thread {
    private static final long LOOK_UP_FREQ = new Long(System.getProperty(
                P2PConstants.PROPERTY_LOOKUP_FREQ)).longValue();
    private static final int MAX_NODE = P2PConstants.MAX_NODE;
    private VirtualNode vn;
    private UniversalService service;
    private VirtualMachine vm;
    private ProActiveRuntime localRuntime;
    int nodeCount = 0;
    long timeout = 0;
    int nodeRequested;
    public static Logger loggerDeployment = ProActiveLogger.getLogger(Loggers.DEPLOYMENT);
    private static final long TIMEOUT = Long.parseLong(System.getProperty(
                P2PConstants.PROPERTY_NODES_ACQUISITION_T0));
    private long expirationTime;

    public ServiceThread(VirtualNode vn, VirtualMachine vm) {
        this.vn = vn;
        this.service = vm.getService();
        this.vm = vm;
        this.localRuntime = ProActiveRuntimeImpl.getProActiveRuntime();
    }

    public void run() {
        ProActiveRuntime[] part = null;

        try {
            part = service.startService();
            nodeCount = nodeCount + ((part != null) ? part.length : 0);
            if (part != null) {
                notifyVirtualNode(part);
            }

            if (service.getServiceName()
                           .equals(SchedulerConstants.SCHEDULER_NODE_NAME)) {
                SchedulerLookupService schedulerLookupService = ((SchedulerLookupService) service);
                Node[] reservedNodes = schedulerLookupService.getNodes();

                for (int i = 0; i < reservedNodes.length; i++) {
                    Node node = reservedNodes[i];
                    nodeCount++;
                    ((VirtualNodeImpl) vn).nodeCreated(new NodeCreationEvent(
                            vn, NodeCreationEvent.NODE_CREATED, node, nodeCount));
                    //                    System.out.println("--------------------" + node.getNodeInformation().getURL());
                    if (loggerDeployment.isInfoEnabled()) {
                        loggerDeployment.info(
                            "Service thread just created event for node: " +
                            node.getNodeInformation().getURL());
                    }
                }
            }

            if (service.getServiceName().equals(P2PConstants.P2P_NODE_NAME)) {
                // Start asking nodes
                P2PService p2pService = ((P2PDescriptorService) service).getP2PService();
                String nodeFamilyRegexp = ((P2PDescriptorService) service).getNodeFamilyRegexp();
                nodeFamilyRegexp = (nodeFamilyRegexp != null)
                    ? nodeFamilyRegexp : "";
                P2PNodeLookup p2pNodesLookup = p2pService.getNodes(((P2PDescriptorService) service).getNodeNumber(),
                        nodeFamilyRegexp, this.vn.getName(), this.vn.getJobID());
                ((VirtualNodeImpl) vn).addP2PNodesLookup(p2pNodesLookup);
                this.nodeRequested = ((P2PDescriptorService) service).getNodeNumber();
                // Timeout
                long vnTimeout = vn.getTimeout();
                this.expirationTime = System.currentTimeMillis() + TIMEOUT;
                if (this.nodeRequested == MAX_NODE) {
                    this.expirationTime = Long.MAX_VALUE;
                    ((VirtualNodeImpl) vn).setTimeout(this.expirationTime, false);
                } else if (vnTimeout < TIMEOUT) {
                    ((VirtualNodeImpl) vn).setTimeout(TIMEOUT, false);
                }

                long step = 100;
                while (askForNodes() &&
                        ((nodeRequested == MAX_NODE) ? true
                                                         : (System.currentTimeMillis() < this.expirationTime))) {
                    if (step > LOOK_UP_FREQ) {
                        step = LOOK_UP_FREQ;
                    }

                    Vector nodes;
                    try {
                        Vector future = p2pNodesLookup.getAndRemoveNodes();
                        nodes = (Vector) ProActive.getFutureValue(future);
                    } catch (Exception e) {
                        loggerDeployment.debug("Couldn't contact the lookup", e);
                        continue;
                    }
                    for (int i = 0; i < nodes.size(); i++) {
                        Node node = (Node) nodes.get(i);
                        nodeCount++;
                        ((VirtualNodeImpl) vn).nodeCreated(new NodeCreationEvent(
                                vn, NodeCreationEvent.NODE_CREATED, node,
                                nodeCount));
                        if (loggerDeployment.isInfoEnabled()) {
                            loggerDeployment.info(
                                "Service thread just created event for node: " +
                                node.getNodeInformation().getURL());
                        }
                    }

                    // Sleeping with FastStart algo
                    if ((this.nodeRequested == MAX_NODE) &&
                            (nodes.size() == 0) && (this.nodeCount != 0)) {
                        Thread.sleep(LOOK_UP_FREQ);
                    } else if (askForNodes() && (this.nodeCount == 0)) {
                        // still no node
                        Thread.sleep(step);
                        step += 100;
                    } else {
                        // normal waiting
                        if (step > LOOK_UP_FREQ) {
                            step = LOOK_UP_FREQ;
                            Thread.sleep(LOOK_UP_FREQ);
                        } else {
                            step += 100;
                            Thread.sleep(step);
                        }
                    }
                }
            }
        } catch (ProActiveException e) {
            loggerDeployment.error(
                "An exception occured while starting the service " +
                service.getServiceName() + " for the VirtualNode " +
                vn.getName() + " \n" + e.getMessage());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void notifyVirtualNode(ProActiveRuntime[] part) {
        for (int i = 0; i < part.length; i++) {
            String url = part[i].getURL();
            String protocol = UrlBuilder.getProtocol(url);
            RuntimeRegistrationEvent event = new RuntimeRegistrationEvent(localRuntime,
                    RuntimeRegistrationEvent.RUNTIME_ACQUIRED, part[i],
                    vn.getName(), protocol, vm.getName());
            ((RuntimeRegistrationEventListener) vn).runtimeRegistered(event);
        }
    }

    private boolean askForNodes() {
        if (nodeRequested == MAX_NODE) {
            // nodeRequested = -1 means try to get the max number of nodes
            return true;
        }
        return nodeCount < nodeRequested;
    }
}

---