org/objectweb/proactive/scheduler/Scheduler.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.scheduler;

import java.util.HashMap;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.objectweb.proactive.ProActive;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeFactory;
import org.objectweb.proactive.core.runtime.ProActiveRuntime;
import org.objectweb.proactive.core.runtime.RuntimeFactory;
import org.objectweb.proactive.core.util.log.Loggers;
import org.objectweb.proactive.core.util.log.ProActiveLogger;
import org.objectweb.proactive.core.util.wrapper.BooleanWrapper;
import org.objectweb.proactive.core.util.wrapper.StringMutableWrapper;
import org.objectweb.proactive.scheduler.policy.*;


public class Scheduler implements java.io.Serializable, SchedulerConstants {
    private static Logger logger = ProActiveLogger.getLogger(Loggers.SCHEDULER);
    static private Scheduler scheduler; // singleton
    private AbstractPolicy policy;
    private RessourceManager ressourceManager;
    private HashMap tmpJobs;

    public Scheduler() {
    }

    public Scheduler(String policyClass) {
        try {
            String nodeURL = "//localhost:" +
                System.getProperty("proactive.rmi.port") + "/" +
                SCHEDULER_NODE_NAME;
            this.tmpJobs = new HashMap();
            this.ressourceManager = (RessourceManager) ProActive.newActive(RessourceManager.class.getName(),
                    new Object[] { new BooleanWrapper(true) }, nodeURL);

            Object[] constructorParameters = new Object[1];
            constructorParameters[0] = this.ressourceManager;
            this.policy = (AbstractPolicy) ProActive.newActive(policyClass,
                    constructorParameters, nodeURL);

            Scheduler.scheduler = this;
            logger.debug("Scheduler created ...");
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public void init() {
    }

    public BooleanWrapper sub(GenericJob job) {
        logger.debug("sub method evoked ....");

        return this.policy.sub(job);
    }

    public BooleanWrapper del(String jobId) {
        logger.debug("del method evoked ....");

        return this.policy.del(jobId);
    }

    public Vector stat(String jobId) {
        logger.debug("stat method evoked ....");
        return this.policy.stat(jobId);
    }

    public Vector nodes(String nodeURL) {
        logger.debug("nodes method evoked ....");
        return this.ressourceManager.nodes(nodeURL);
    }

    static public void createScheduler(String policyName) {
        // tests if the service already exists
        String nodeURL = System.getProperty(SCHEDULER_URL) + "/" +
            SCHEDULER_NODE_NAME;
        Scheduler.scheduler = Scheduler.connectTo(nodeURL);

        if (Scheduler.scheduler == null) {
            //                  JVMProcessImpl rsh = new JVMProcessImpl(new StandardOutputMessageLogger());
            //          rsh.setClassname("org.objectweb.proactive.core.node.StartNode");
            //          rsh.setParameters(nodeURL);
            //          rsh.startProcess();                             

            //          Thread.sleep(3000);

            //String nodeURL = "//localhost/" + SCHEDULER_NODE_NAME;
            ProActiveRuntime paRuntime;

            try {
                paRuntime = RuntimeFactory.getProtocolSpecificRuntime(System.getProperty(
                            "proactive.communication.protocol") + ":");
                nodeURL = paRuntime.createLocalNode(SCHEDULER_NODE_NAME, false,
                        null, paRuntime.getVMInformation().getName(),
                        paRuntime.getJobID());

                Object[] constructorParameters = { policyName };
                scheduler = (Scheduler) ProActive.newActive(Scheduler.class.getName(),
                        constructorParameters, nodeURL);

                logger.debug("created object scheduler");
                //scheduler.init();
            } catch (Exception e) {
                // TODO Auto-generated catch block
                logger.error("error creating object scheduler");
            }
        }
    }

    public static void start(String policyName) {
        try {
            Scheduler.createScheduler(policyName);
            logger.debug("service started successfully");
        } catch (Exception e) {
            // TODO Auto-generated catch block
            logger.error("error starting service");
        }
    }

    public void end() {
        logger.debug("shutting down scheduler service");
        policy.end();
    }

    public StringMutableWrapper info() {
        return new StringMutableWrapper("Scheduler beta");
    }

    protected void finalize() {
        this.end();
        Scheduler.scheduler = null;
    }

    public StringMutableWrapper fetchJobDescription(String xmlDescriptorUrl) {
        try {
            // it is here that we shall enter the modifications if the submission of the 
            // xml file is remote...
            GenericJob tmp = new GenericJob();
            String xmlFileName = xmlDescriptorUrl.substring(xmlDescriptorUrl.lastIndexOf(
                        '/') + 1);
            String jobId = tmp.getJobID();
            tmp.setXMLDescriptorName(xmlFileName);
            tmp.setXMLFullPath(xmlDescriptorUrl);
            this.tmpJobs.put(jobId, tmp);

            ProActiveJobHandler h = new ProActiveJobHandler(this, jobId,
                    xmlDescriptorUrl);
            String uri = xmlDescriptorUrl;
            org.objectweb.proactive.core.xml.io.StreamReader sr = new org.objectweb.proactive.core.xml.io.StreamReader(new org.xml.sax.InputSource(
                        uri), h);
            sr.read();

            logger.debug("starting the parsing of the newly added XML file");

            return new StringMutableWrapper(jobId);
        } catch (Exception e) {
            // log it for futur reference ...
            logger.error("error parsing the file");
            return null;
        }
    }

    public static Scheduler connectTo(String schedulerURL) {
        try {
            Node node = NodeFactory.getNode(schedulerURL);

            Object[] ao = node.getActiveObjects(Scheduler.class.getName());

            if (ao.length == 1) {
                logger.debug("scheduler object fetched");
                return ((Scheduler) ao[0]);
            } else {
                logger.debug("no scheduler object created");
                return null;
            }
        } catch (Exception e) {
            logger.error("error connecting the scheduler service:" + e);
            return null;
        }
    }

    public static void main(String[] args) throws Exception {
        String policyName = "org.objectweb.proactive.scheduler.policy.FIFOPolicy";
        Scheduler.start(policyName);
        Thread.sleep(3000);
    }

    public Vector getNodes(int ressourceNb, int estimatedTime) {
        Object[] constructorParameters = new Object[3];
        constructorParameters[0] = this.ressourceManager;
        constructorParameters[1] = new Integer(ressourceNb);
        constructorParameters[2] = new Integer(estimatedTime);
        try {
            JobNoDescriptor job = (JobNoDescriptor) ProActive.newActive(JobNoDescriptor.class.getName(),
                    constructorParameters);
            GenericJob jobDescription = job.getJobDescription();
            this.sub(jobDescription);
            logger.debug("new object created in the queue");
            return job.getNodes();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            logger.error("error submitting the command to the scheduler");

            return null;
        }
    }

    // ------------------
    // methods used for parsing .....

    public Node[] getReservedNodes(String jobID, int askedNodes) {
        logger.debug("returning reserved nodes");
        return this.ressourceManager.getNodes(jobID, askedNodes);
    }

    public void commit(String jobID) {
        if (this.tmpJobs.containsKey(jobID)) {
            System.out.println("committing the job's description");

            GenericJob job = (GenericJob) this.tmpJobs.remove(jobID);
            this.sub(job);
            logger.debug("commiting the job's description");
        } else {
            logger.debug("no object to commit");
        }
    }

    public GenericJob getTmpJob(String jobID) {
        if (this.tmpJobs.containsKey(jobID)) {
            return (GenericJob) this.tmpJobs.get(jobID);
        } else {
            return null;
        }
    }
}

---