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

import java.io.IOException;
import java.net.MalformedURLException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;

import org.apache.log4j.Logger;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.UniqueID;
import org.objectweb.proactive.core.body.AbstractBody;
import org.objectweb.proactive.core.body.UniversalBody;
import org.objectweb.proactive.core.body.ft.checkpointing.CheckpointInfo;
import org.objectweb.proactive.core.body.ft.internalmsg.FTMessage;
import org.objectweb.proactive.core.body.ft.internalmsg.Heartbeat;
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.storage.CheckpointServer;
import org.objectweb.proactive.core.body.reply.Reply;
import org.objectweb.proactive.core.body.request.Request;
import org.objectweb.proactive.core.config.ProActiveConfiguration;
import org.objectweb.proactive.core.util.log.Loggers;
import org.objectweb.proactive.core.util.log.ProActiveLogger;
import org.objectweb.proactive.ext.security.exceptions.RenegotiateSessionException;


public abstract class FTManager implements java.io.Serializable {
    //logger
    protected static Logger logger = ProActiveLogger.getLogger(Loggers.FAULT_TOLERANCE);

    public static final int NON_FT = -30;

    public static final int DEFAULT_TTC_VALUE = 30000;

    public static final int IMMEDIATE_SERVICE = -1;

    public static final int ORPHAN_REPLY = -2;

    public static final long TIME_TO_RESEND = 3000;

    public static final String HALF_BODY_EXCEPTION_MESSAGE = "Cannot perform this call on a FTManager of a HalfBody";

    // true is this is a checkpoint
    private boolean isACheckpoint;

    // body attached to this manager
    protected AbstractBody owner;
    protected UniqueID ownerID;

    // server adresses
    protected CheckpointServer storage;
    protected LocationServer location;
    protected RecoveryProcess recovery;

    // additional codebase for checkpoints
    protected String additionalCodebase;

    // checkpoint interval (ms)
    protected int ttc;

    public static int getProtoSelector(String protoName) {
        if ("cic".equals(protoName)) {
            return FTManagerFactory.PROTO_CIC;
        } else if ("pml".equals(protoName)) {
            return FTManagerFactory.PROTO_PML;
        }
        return 0;
    }

    public int init(AbstractBody owner) throws ProActiveException {
        this.owner = owner;
        this.ownerID = owner.getID();
        try {
            String ttcValue = ProActiveConfiguration.getTTCValue();
            if (ttcValue != null) {
                this.ttc = Integer.parseInt(ttcValue) * 1000;
            } else {
                this.ttc = FTManager.DEFAULT_TTC_VALUE;
            }
            String urlGlobal = ProActiveConfiguration.getGlobalFTServer();
            if (urlGlobal != null) {
                this.storage = (CheckpointServer) (Naming.lookup(urlGlobal));
                this.location = (LocationServer) (Naming.lookup(urlGlobal));
                this.recovery = (RecoveryProcess) (Naming.lookup(urlGlobal));
            } else {
                String urlCheckpoint = ProActiveConfiguration.getCheckpointServer();
                String urlRecovery = ProActiveConfiguration.getRecoveryServer();
                String urlLocation = ProActiveConfiguration.getLocationServer();
                if ((urlCheckpoint != null) && (urlRecovery != null) &&
                        (urlLocation != null)) {
                    this.storage = (CheckpointServer) (Naming.lookup(urlCheckpoint));
                    this.location = (LocationServer) (Naming.lookup(urlLocation));
                    this.recovery = (RecoveryProcess) (Naming.lookup(urlRecovery));
                } else {
                    throw new ProActiveException(
                        "Unable to init FTManager : servers are not correctly set");
                }
            }

            // the additional codebase is added to normal codebase 
            // ONLY during serialization for checkpoint !
            this.additionalCodebase = this.storage.getServerCodebase();

            // registration in the recovery process and in the localisation server
            try {
                this.recovery.register(ownerID);
                this.location.updateLocation(ownerID, owner.getRemoteAdapter());
            } catch (RemoteException e) {
                logger.error("**ERROR** Unable to register in location server");
                throw new ProActiveException("Unable to register in location server",
                    e);
            }
        } catch (MalformedURLException e) {
            throw new ProActiveException("Unable to init FTManager : FT is disable.",
                e);
        } catch (RemoteException e) {
            throw new ProActiveException("Unable to init FTManager : FT is disable.",
                e);
        } catch (NotBoundException e) {
            throw new ProActiveException("Unable to init FTManager : FT is disable.",
                e);
        }
        return 0;
    }

    public void termination() throws ProActiveException {
        try {
            this.recovery.unregister(this.ownerID);
        } catch (RemoteException e) {
            logger.error("**ERROR** Unable to register in location server");
            throw new ProActiveException("Unable to unregister in location server",
                e);
        }
    }

    public boolean isACheckpoint() {
        return isACheckpoint;
    }

    public void setCheckpointTag(boolean tag) {
        this.isACheckpoint = tag;
    }

    public UniversalBody communicationFailed(UniqueID suspect,
        UniversalBody suspectLocation, Exception e) {
        try {
            // send an adapter to suspectLocation: the suspected body could be local
            UniversalBody newLocation = this.location.searchObject(suspect,
                    suspectLocation.getRemoteAdapter(), this.ownerID);
            if (newLocation == null) {
                while (newLocation == null) {
                    try {
                        // suspected is failed or is recovering
                        if (logger.isDebugEnabled()) {
                            logger.debug("[CIC] Waiting for recovery of " +
                                suspect);
                        }
                        Thread.sleep(TIME_TO_RESEND);
                    } catch (InterruptedException e2) {
                        e2.printStackTrace();
                    }
                    newLocation = this.location.searchObject(suspect,
                            suspectLocation.getRemoteAdapter(), this.ownerID);
                }
                return newLocation;
            } else {
                // newLocation is the new location of suspect
                return newLocation;
            }
        } catch (RemoteException e1) {
            logger.error("**ERROR** Location server unreachable");
            e1.printStackTrace();
            return null;
        }
    }

    public int sendReply(Reply r, UniversalBody destination) {
        try {
            this.onSendReplyBefore(r);
            int res = r.send(destination);
            this.onSendReplyAfter(r, res, destination);
            return res;
        } catch (IOException e) {
            logger.info("[FAULT] " + this.ownerID + " : FAILURE OF " +
                destination.getID() + " SUSPECTED ON REPLY SENDING : " +
                e.getMessage());
            UniversalBody newDestination = this.communicationFailed(destination.getID(),
                    destination, e);
            return this.sendReply(r, newDestination);
        }
    }

    public int sendRequest(Request r, UniversalBody destination)
        throws RenegotiateSessionException {
        try {
            this.onSendRequestBefore(r);
            int res = r.send(destination);
            this.onSendRequestAfter(r, res, destination);
            return res;
        } catch (IOException e) {
            logger.info("[FAULT] " + this.ownerID + " : FAILURE OF " +
                destination.getID() + " SUSPECTED ON REQUEST SENDING : " +
                e.getMessage());
            UniversalBody newDestination = this.communicationFailed(destination.getID(),
                    destination, e);
            return this.sendRequest(r, newDestination);
        } catch (RenegotiateSessionException e1) {
            throw e1;
        }
    }

    public Object handleHBEvent(Heartbeat fte) {
        if (this.owner.isAlive()) {
            return FaultDetector.OK;
        } else {
            return FaultDetector.IS_DEAD;
        }
    }

    // ABSTRACT METHODS //

    public abstract int onReceiveReply(Reply reply);

    public abstract int onReceiveRequest(Request request);

    public abstract int onDeliverReply(Reply reply);

    public abstract int onDeliverRequest(Request request);

    public abstract int onSendReplyBefore(Reply reply);

    public abstract int onSendReplyAfter(Reply reply, int rdvValue,
        UniversalBody destination);

    public abstract int onSendRequestBefore(Request request);

    public abstract int onSendRequestAfter(Request request, int rdvValue,
        UniversalBody destination) throws RenegotiateSessionException;

    public abstract int onServeRequestBefore(Request request);

    public abstract int onServeRequestAfter(Request request);

    public abstract int beforeRestartAfterRecovery(CheckpointInfo ci, int inc);

    public abstract Object handleFTMessage(FTMessage fte);
    
    public void updateLocationAtServer(UniqueID ownerID, UniversalBody remoteBodyAdapter) {}
}

---