spmddistributedexecutionmanager.jad

collective processing environment for mobile devices. It is an environment for mobile device to solv

// Decompiled by DJ v2.6.6.51 Copyright 2000 Atanas Neshkov  Date: 19/06/2006 9:50:24 PM
// Home Page : http://members.fortunecity.com/neshkov/dj.html  - Check often for new version!
// Decompiler options: packimports(3) 
// Source File Name:   SPMDDistributedExecutionManager.java

package com.ibm.dthreads;

import java.io.FileInputStream;
import java.io.PrintStream;
import java.rmi.Naming;
import java.rmi.RemoteException;
import java.util.*;

// Referenced classes of package com.ibm.dthreads:
//            DistributedExecutionManager, DaemonRegistry, DThreadExecutionContext, DistributedExecutionDaemon, 
//            DThreadContractor, DThread

public class SPMDDistributedExecutionManager
    implements DistributedExecutionManager
{

    public SPMDDistributedExecutionManager(int managerId)
    {
        servicenamearray = null;
        daemonslist = null;
        noofdaemons = 0;
        serverProperties = new Properties();
        try
        {
            FileInputStream serverpropsfile = new FileInputStream("daemonregistry.properties");
            this.managerId = managerId;
            serverProperties.load(serverpropsfile);
            hostName = (String)serverProperties.get("daemonregistry");
            serviceName = (String)serverProperties.get("remoteservicename");
            daemonRegistry = (DaemonRegistry)Naming.lookup("//" + hostName + "/" + serviceName);
        }
        catch(Exception e)
        {
            System.out.println("STMDDistributedExecutionManager::constructor() Exception" + e.getMessage());
        }
    }

    public String execute(DThread dthread, DThreadExecutionContext context)
    {
        noofdaemons = context.getNumberOfThreads();
        String ipList[] = new String[noofdaemons];
        debugMode = context.checkdebugMode();
        daemonarray = new DistributedExecutionDaemon[noofdaemons];
        servicenamearray = new String[noofdaemons];
        dthreadcontractorarray = new DThreadContractor[noofdaemons];
        try
        {
            daemonslist = daemonRegistry.reserveDaemons(noofdaemons, managerId);
            if(daemonslist.size() <= 0)
            {
                System.out.println("STMDDistributedExecutionManager::execute() Not enough Daemons to run");
                for(int i = 0; i < noofdaemons; i++)
                    daemonRegistry.releaseDaemon(servicenamearray[i]);

                return null;
            }
        }
        catch(RemoteException e)
        {
            System.out.println("STMDDistributedExecutionManager::execute():Exception is " + e.getStackTrace());
            e.printStackTrace();
        }
        try
        {
            servicenamearray = daemonRegistry.getServicenames(managerId);
        }
        catch(RemoteException e1)
        {
            System.out.println("STMDDistributedExecutionManager::execute():Exception is " + e1.getStackTrace());
            e1.printStackTrace();
        }
        for(int iter = 0; iter < noofdaemons; iter++)
        {
            daemonarray[iter] = (DistributedExecutionDaemon)daemonslist.get(iter);
            try
            {
                if(noofdaemons == 1)
                {
                    StringTokenizer findRoothost = new StringTokenizer(servicenamearray[iter], "//");
                    String ipAddress = findRoothost.nextToken();
                    ipList[iter] = ipAddress;
                } else
                {
                    StringTokenizer findRoothost = new StringTokenizer(servicenamearray[iter], "//");
                    String ipAddress = findRoothost.nextToken();
                    ipList[iter] = ipAddress;
                }
            }
            catch(Exception e)
            {
                System.out.println("Exception in SPMDDistributedExecutionManager.execute() ");
                e.printStackTrace();
            }
            try
            {
                daemonarray[iter].init(noofdaemons, iter, context.getexperimentName());
            }
            catch(RemoteException e2)
            {
                System.out.println("STMDDistributedExecutionManager::execute(): Exception is " + e2.getStackTrace());
                System.out.println(" DAEMON " + servicenamearray[iter] + " IS DEAD ");
                System.out.println("GETTING ADDITIONAL DAEMON");
                try
                {
                    daemonRegistry.unregisterDaemon(servicenamearray[iter]);
                    daemonarray[iter] = daemonRegistry.getAdditionaldaemon();
                    if(daemonarray[iter] == null)
                    {
                        System.out.println("SORRY ! [INSIDE STMD:] NOT ENOUGH DAEMONS");
                        for(int i = 0; i < noofdaemons; i++)
                            daemonRegistry.releaseDaemon(servicenamearray[i]);

                        return null;
                    }
                    servicenamearray[iter] = daemonRegistry.getAdditionaldaemonName();
                    daemonarray[iter].init(noofdaemons, iter, context.getexperimentName());
                }
                catch(RemoteException e22)
                {
                    System.out.println("STMDDistributedExecutionManager::execute():Exception is " + e22.getStackTrace());
                    System.out.println("IN GETTING ADDITIONAL DAEMON");
                    e22.printStackTrace();
                }
            }
            dthreadcontractorarray[iter] = new DThreadContractor();
        }

        Thread masterthread = null;
        for(int iter = 0; iter < noofdaemons; iter++)
        {
            DThreadExecutionContext exc = new DThreadExecutionContext(noofdaemons);
            if(debugMode)
                exc.setdebugMode();
            exc.setIdentity(iter);
            exc.setAssociatedDaemons(daemonslist);
            exc.setDaemonarray(daemonarray);
            dthread.setContext(exc);
            exc.setDthreadobj(dthread);
            DThread newthread = (DThread)dthread.getClone();
            System.out.println("DThread " + DThread.getContext().getIdentity() + " has started execution");
            dthreadcontractorarray[iter].setfields(daemonarray[iter], newthread, daemonRegistry, servicenamearray[iter]);
            try
            {
                dthreadcontractorarray[iter].start();
            }
            catch(Exception e)
            {
                System.out.println("GETTING ADDITIONAL DAEMON");
                try
                {
                    daemonarray[iter] = daemonRegistry.getAdditionaldaemon();
                    servicenamearray[iter] = daemonRegistry.getAdditionaldaemonName();
                }
                catch(RemoteException e2)
                {
                    System.out.println("STMDDistributedExecutionManager::execute():Exception is " + e2.getStackTrace());
                    e2.printStackTrace();
                }
                dthreadcontractorarray[iter].setfields(daemonarray[iter], newthread, daemonRegistry, servicenamearray[iter]);
                dthreadcontractorarray[iter].start();
            }
            if(iter == 0)
                masterthread = dthreadcontractorarray[iter];
        }

        if(masterthread != null)
            try
            {
                masterthread.join();
            }
            catch(InterruptedException e2)
            {
                System.out.println("STMDDistributedExecutionManager::execute():Exception is " + e2.getStackTrace());
                e2.printStackTrace();
            }
        return ipList[0];
    }

    private Properties serverProperties;
    private String hostName;
    private String serviceName;
    private DaemonRegistry daemonRegistry;
    String servicenamearray[];
    ArrayList daemonslist;
    int noofdaemons;
    DistributedExecutionDaemon daemonarray[];
    DThreadContractor dthreadcontractorarray[];
    boolean debugMode;
    int managerId;
}