org/objectweb/proactive/core/body/ft/servers/FTServer.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.body.ft.servers;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;

import org.apache.log4j.Logger;
import org.objectweb.proactive.core.UniqueID;
import org.objectweb.proactive.core.body.UniversalBody;
import org.objectweb.proactive.core.body.ft.checkpointing.Checkpoint;
import org.objectweb.proactive.core.body.ft.checkpointing.CheckpointInfo;
import org.objectweb.proactive.core.body.ft.message.HistoryUpdater;
import org.objectweb.proactive.core.body.ft.message.MessageInfo;
import org.objectweb.proactive.core.body.ft.servers.faultdetection.FaultDetector;
import org.objectweb.proactive.core.body.ft.servers.location.LocationServer;
import org.objectweb.proactive.core.body.ft.servers.recovery.RecoveryProcess;
import org.objectweb.proactive.core.body.ft.servers.resource.ResourceServer;
import org.objectweb.proactive.core.body.ft.servers.storage.CheckpointServer;
import org.objectweb.proactive.core.body.ft.servers.util.ActiveQueueJob;
import org.objectweb.proactive.core.body.ft.servers.util.JobBarrier;
import org.objectweb.proactive.core.body.reply.Reply;
import org.objectweb.proactive.core.body.request.Request;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.util.log.Loggers;
import org.objectweb.proactive.core.util.log.ProActiveLogger;


public class FTServer extends UnicastRemoteObject implements FaultDetector,
    LocationServer, RecoveryProcess, ResourceServer, CheckpointServer {
    //logger
    protected static Logger logger = ProActiveLogger.getLogger(Loggers.FAULT_TOLERANCE);

    public static final int DEFAULT_PORT = 1100;

    public static final int DEFAULT_FDETECT_SCAN_PERIOD = 10000;

    public static final String DEFAULT_SERVER_NAME = "FTServer";

    // internal servers
    private FaultDetector faultDetector;
    private LocationServer locationServer;
    private RecoveryProcess recoveryProcess;
    private ResourceServer resourceServer;
    private CheckpointServer checkpointServer;

    public FTServer() throws RemoteException {
        super();
        // this.killingQueue.start();
    }

    public void init(FaultDetector fd, LocationServer ls, RecoveryProcess rp,
        ResourceServer rs, CheckpointServer cs) {
        this.faultDetector = fd;
        this.locationServer = ls;
        this.recoveryProcess = rp;
        this.resourceServer = rs;
        this.checkpointServer = cs;
    }


    public boolean isUnreachable(UniversalBody body) throws RemoteException {
        return this.faultDetector.isUnreachable(body);
    }

    public void startFailureDetector() throws RemoteException {
        this.faultDetector.startFailureDetector();
    }

    public void suspendFailureDetector() throws RemoteException {
        this.faultDetector.suspendFailureDetector();
    }

    public void stopFailureDetector() throws RemoteException {
        this.faultDetector.stopFailureDetector();
    }

    public void forceDetection() throws RemoteException {
        this.faultDetector.forceDetection();
    }

    public UniversalBody searchObject(UniqueID id, UniversalBody oldLocation,
        UniqueID caller) throws RemoteException {
        return this.locationServer.searchObject(id, oldLocation, caller);
    }

    public void updateLocation(UniqueID id, UniversalBody newLocation)
        throws RemoteException {
        this.locationServer.updateLocation(id, newLocation);
    }

    public ArrayList getAllLocations() throws RemoteException {
        return this.locationServer.getAllLocations();
    }

    public UniversalBody getLocation(UniqueID id) throws RemoteException {
        return this.locationServer.getLocation(id);
    }

    public void register(UniqueID id) throws RemoteException {
        this.recoveryProcess.register(id);
    }

    public void unregister(UniqueID id) throws RemoteException {
        this.recoveryProcess.unregister(id);
    }

    public void failureDetected(UniqueID id) throws RemoteException {
        this.recoveryProcess.failureDetected(id);
    }

    public void updateState(UniqueID id, int state) throws RemoteException {
        this.recoveryProcess.updateState(id, state);
    }

    public int getSystemSize() throws RemoteException {
        return this.recoveryProcess.getSystemSize();
    }

    public void submitJob(ActiveQueueJob job) throws RemoteException {
        this.recoveryProcess.submitJob(job);
    }

    public JobBarrier submitJobWithBarrier(ActiveQueueJob job)
        throws RemoteException {
        return this.recoveryProcess.submitJobWithBarrier(job);
    }

    public void addFreeNode(Node n) throws RemoteException {
        this.resourceServer.addFreeNode(n);
    }

    public Node getFreeNode() throws RemoteException {
        return this.resourceServer.getFreeNode();
    }

    // Commented code is used to benchmark recovery time. A failure is triggered
    // artifially by sending a kill message to one active object.
    //public static final int numCkpt = 4; // totalWorker * failureFrequency
    //public static final int numberOfWorker = 4;
    //public static final long timeToWait = 45000; // TTC/2
    //public static final int totalFailure = 1;
    //private int ckptCounter = 1;
    //private int failureCounter = 1;
    //private int firstCkptCounter = numberOfWorker;
    //private ActiveQueue killingQueue = new ActiveQueue("Killing thread");
    public int storeCheckpoint(Checkpoint c, int incarnation)
        throws RemoteException {
        //this.firstCkptCounter--;
        //if (this.firstCkptCounter < 0){
        //    if ((this.failureCounter <= totalFailure) && 
        //            ((ckptCounter) == numCkpt)) {
        //        this.killingQueue.addJob(
        //                new KillerJob(this,this.getLocation(c.getBodyID()),timeToWait));
        //        this.ckptCounter=0; // the following ++ set it to 1 ...
        //        this.failureCounter++;
        //    }
        //    this.ckptCounter++;
        //}
        return this.checkpointServer.storeCheckpoint(c, incarnation);
    }

    public Checkpoint getCheckpoint(UniqueID id, int sequenceNumber)
        throws RemoteException {
        return this.checkpointServer.getCheckpoint(id, sequenceNumber);
    }

    public Checkpoint getLastCheckpoint(UniqueID id) throws RemoteException {
        return this.checkpointServer.getLastCheckpoint(id);
    }

    public void addInfoToCheckpoint(CheckpointInfo ci, UniqueID id,
        int sequenceNumber, int incarnation) throws RemoteException {
        this.checkpointServer.addInfoToCheckpoint(ci, id, sequenceNumber,
            incarnation);
    }

    public CheckpointInfo getInfoFromCheckpoint(UniqueID id, int sequenceNumber)
        throws RemoteException {
        return this.checkpointServer.getInfoFromCheckpoint(id, sequenceNumber);
    }

    public void commitHistory(HistoryUpdater rh) throws RemoteException {
        this.checkpointServer.commitHistory(rh);
    }

    public void outputCommit(MessageInfo mi) throws RemoteException {
        this.checkpointServer.outputCommit(mi);
    }

    public String getServerCodebase() throws RemoteException {
        return this.checkpointServer.getServerCodebase();
    }

    public void storeRequest(UniqueID receiverId, Request request)
        throws RemoteException {
        this.checkpointServer.storeRequest(receiverId, request);
    }

    public void storeReply(UniqueID receiverID, Reply reply)
        throws RemoteException {
        this.checkpointServer.storeReply(receiverID, reply);
    }


    public CheckpointServer getCheckpointServer() {
        return checkpointServer;
    }

    public void setCheckpointServer(CheckpointServer checkpointServer) {
        this.checkpointServer = checkpointServer;
    }

    public FaultDetector getFaultDetector() {
        return faultDetector;
    }

    public void setFaultDetector(FaultDetector faultDetector) {
        this.faultDetector = faultDetector;
    }

    public LocationServer getLocationServer() {
        return locationServer;
    }

    public void setLocationServer(LocationServer locationServer) {
        this.locationServer = locationServer;
    }

    public RecoveryProcess getRecoveryProcess() {
        return recoveryProcess;
    }

    public void setRecoveryProcess(RecoveryProcess recoveryProcess) {
        this.recoveryProcess = recoveryProcess;
    }

    public ResourceServer getResourceServer() {
        return resourceServer;
    }

    public void setResourceServer(ResourceServer resourceServer) {
        this.resourceServer = resourceServer;
    }

    public void initialize() throws RemoteException {
        logger.info("[GLOBAL] Reinitializing server ...");
        this.checkpointServer.initialize();
        this.locationServer.initialize();
        this.recoveryProcess.initialize();
        this.resourceServer.initialize();
        logger.info("[GLOBAL] Server reinitialized");
    }
}

---