org/objectweb/proactive/core/body/future/FutureProxy.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.future;

import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;

import org.objectweb.proactive.Body;
import org.objectweb.proactive.ProActive;
import org.objectweb.proactive.core.Constants;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.ProActiveRuntimeException;
import org.objectweb.proactive.core.UniqueID;
import org.objectweb.proactive.core.body.LocalBodyStore;
import org.objectweb.proactive.core.body.UniversalBody;
import org.objectweb.proactive.core.body.proxy.AbstractProxy;
import org.objectweb.proactive.core.event.FutureEvent;
import org.objectweb.proactive.core.exceptions.manager.ExceptionHandler;
import org.objectweb.proactive.core.exceptions.manager.ExceptionMaskLevel;
import org.objectweb.proactive.core.exceptions.manager.NFEManager;
import org.objectweb.proactive.core.exceptions.proxy.FutureTimeoutException;
import org.objectweb.proactive.core.exceptions.proxy.ProxyNonFunctionalException;
import org.objectweb.proactive.core.mop.ConstructionOfReifiedObjectFailedException;
import org.objectweb.proactive.core.mop.ConstructorCall;
import org.objectweb.proactive.core.mop.MOP;
import org.objectweb.proactive.core.mop.MethodCall;
import org.objectweb.proactive.core.mop.MethodCallExecutionFailedException;
import org.objectweb.proactive.core.mop.Proxy;
import org.objectweb.proactive.core.mop.StubObject;


public class FutureProxy implements Future, Proxy, java.io.Serializable {
    //
    // -- STATIC MEMBERS -----------------------------------------------
    //

    public static final int RECYCLE_POOL_SIZE = 1000;
    private static FutureProxy[] recyclePool;

    private static boolean shouldPoolFutureProxyObjects;
    private static int index;

    private static FutureEventProducerImpl futureEventProducer;

    //
    // -- PROTECTED MEMBERS -----------------------------------------------
    //

    protected FutureResult target;

    protected boolean migration;

    protected boolean continuation;

    protected UniqueID creatorID;

    protected long ID;

    protected UniqueID senderID;

    protected HashMap futureLevel = null;

    private ExceptionMaskLevel exceptionLevel;

    private transient AbstractProxy originatingProxy;

    protected static long futureMaxDelay = -1;

    //
    // -- CONSTRUCTORS -----------------------------------------------
    //

    public FutureProxy() throws ConstructionOfReifiedObjectFailedException {
    }

    public FutureProxy(ConstructorCall c, Object[] p)
        throws ConstructionOfReifiedObjectFailedException {
        // we don't care what the arguments are
        this();
    }

    //
    // -- PUBLIC STATIC METHODS -----------------------------------------------
    //

    public static boolean isAwaited(Object obj) {
        return ProActive.isAwaited(obj);
    }

    public synchronized static FutureProxy getFutureProxy() {
        FutureProxy result;
        if (shouldPoolFutureProxyObjects && (index > 0)) {
            // gets the object from the pool
            index--;
            result = recyclePool[index];
            recyclePool[index] = null;
        } else {
            try {
                result = new FutureProxy();
            } catch (ConstructionOfReifiedObjectFailedException e) {
                result = null;
            }
        }
        return result;
    }

    public static FutureEventProducer getFutureEventProducer() {
        if (futureEventProducer == null) {
            futureEventProducer = new FutureEventProducerImpl();
        }
        return futureEventProducer;
    }

    //
    // -- PUBLIC METHODS -----------------------------------------------
    //
    public boolean equals(Object obj) {
        //we test if we have a future object
        if (isFutureObject(obj)) {
            return (((StubObject) obj).getProxy().hashCode() == this.hashCode());
        }
        return false;
    }

    //
    // -- Implements Future -----------------------------------------------
    //

    public synchronized void receiveReply(FutureResult obj)
        throws java.io.IOException {
        if (isAvailable()) {
            throw new java.io.IOException(
                "FutureProxy receives a reply and this target field is not null");
        }
        target = obj;
        ExceptionHandler.addResult(this);
        ProxyNonFunctionalException nfe = target.getNFE();
        if (nfe != null) {
            NFEManager.fireNFE(nfe, originatingProxy);
        }

        originatingProxy = null;
        this.notifyAll();
    }

    public synchronized Throwable getRaisedException() {
        waitFor();
        return target.getExceptionToRaise();
    }

    public boolean isAvailable() {
        return target != null;
    }

    public FutureResult getFutureResult() {
        return target;
    }

    public synchronized Object getResult() {
        waitFor();
        return target.getResult();
    }

    public synchronized void setResult(Object o) {
        target = new FutureResult(o, null, null);
    }

    public synchronized boolean isAwaited() {
        return !isAvailable();
    }

    public synchronized void waitFor() {
        if (futureMaxDelay == -1) {

            /* First time, hopefully the configuration file has been read */
            try {
                futureMaxDelay = Long.parseLong(System.getProperty(
                            "proactive.future.maxdelay"));
            } catch (IllegalArgumentException iea) {

                /* The property is not set, that's not a problem */
                futureMaxDelay = 0;
            }
        }

        try {
            waitFor(futureMaxDelay);
        } catch (ProActiveException e) {
            ProxyNonFunctionalException nfe = new FutureTimeoutException(
                    "Exception after waiting for " + futureMaxDelay + "ms", e);

            target = new FutureResult(null, null, nfe);
            notifyAll();
        }
    }

    public synchronized void waitFor(long timeout) throws ProActiveException {
        if (isAvailable()) {
            return;
        }

        UniqueID id = null;

        // send WAIT_BY_NECESSITY event to listeners if there are any
        if (futureEventProducer != null) {
            id = ProActive.getBodyOnThis().getID();
            if (LocalBodyStore.getInstance().getLocalBody(id) != null) {
                // send event only if ActiveObject, not for HalfBodies
                futureEventProducer.notifyListeners(id, getCreatorID(),
                    FutureEvent.WAIT_BY_NECESSITY);
            } else {
                id = null;
            }
        }
        int timeoutCounter = 1;
        while (!isAvailable()) {
            timeoutCounter--;
            // counter < 0 means that it is the second time we enter in the loop
            // while still not available, i.e timeout has expired
            if (timeoutCounter < 0) {
                throw new ProActiveException(
                    "Timeout expired while waiting for the future update");
            }
            try {
                this.wait(timeout);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // send RECEIVED_FUTURE_RESULT event to listeners if there are any
        if (id != null) {
            futureEventProducer.notifyListeners(id, getCreatorID(),
                FutureEvent.RECEIVED_FUTURE_RESULT);
        }
    }

    public long getID() {
        return ID;
    }

    public void setID(long l) {
        ID = l;
    }

    public void setCreatorID(UniqueID i) {
        creatorID = i;
    }

    public UniqueID getCreatorID() {
        return creatorID;
    }

    public void setSenderID(UniqueID i) {
        senderID = i;
    }

    public void setOriginatingProxy(AbstractProxy p) {
        originatingProxy = p;
    }

    //
    // -- Implements Proxy -----------------------------------------------
    //

    public Object reify(MethodCall c) throws InvocationTargetException {
        Object result = null;

        //stem.out.println("FutureProxy: c.getName() = " +c.getName());
        //              if ((c.getName()).equals("equals") || (c.getName()).equals("hashCode")) {
        //                      //System.out.println("FutureProxy: now executing " + c.getName());
        //                      try {
        //                              result = c.execute(this);
        //                      } catch (MethodCallExecutionFailedException e) {
        //                              throw new ProActiveRuntimeException("FutureProxy: Illegal arguments in call " + c.getName());
        //                      }
        //                      return result;
        //              }
        waitFor();

        // Now that the object is available, execute the call
        Object resultObject = target.getResult();
        try {
            result = c.execute(resultObject);
        } catch (MethodCallExecutionFailedException e) {
            throw new ProActiveRuntimeException(
                "FutureProxy: Illegal arguments in call " + c.getName());
        }

        // If target of this future is another future, make a shortcut !
        if (resultObject instanceof StubObject) {
            Proxy p = ((StubObject) resultObject).getProxy();
            if (p instanceof FutureProxy) {
                target = ((FutureProxy) p).target;
            }
        }

        return result;
    }

    // -- PROTECTED METHODS -----------------------------------------------
    //
    protected void finalize() {
        returnFutureProxy(this);
    }

    protected void setMigrationTag() {
        migration = true;
    }

    protected void unsetMigrationTag() {
        migration = false;
    }

    public synchronized void setContinuationTag() {
        continuation = true;
    }

    public synchronized void unsetContinuationTag() {
        continuation = false;
    }

    //
    // -- PRIVATE METHODS FOR SERIALIZATION -----------------------------------------------
    //
    private synchronized void writeObject(java.io.ObjectOutputStream out)
        throws java.io.IOException {
        if (!FuturePool.isInsideABodyForwarder()) {
            // If we are on a forwarder we want to forward the call, not wait the 
            // futur result or whatever
            //if continuation is already set, we are in a forwarder
            //else if a destination is available in destTable, set the continuation tag
            if (!continuation) {
                continuation = (FuturePool.getBodyDestination() != null);
            }

            // We wait until the result is available
            if ((!migration) && (!continuation)) {
                waitFor();
            }

            // Registration in case of continuation
            if (continuation && isAwaited()) {
                // get the sender body
                Body sender = LocalBodyStore.getInstance().getLocalBody(senderID);

                // it's a halfbody...
                if (sender == null) {
                    sender = LocalBodyStore.getInstance().getLocalHalfBody(senderID);
                }
                if (sender != null) {
                    UniversalBody dest = FuturePool.getBodyDestination();
                    if (dest != null) {
                        sender.getFuturePool().addAutomaticContinuation(ID,
                            creatorID, dest);
                    }
                }

                // if sender is still null, it's a forwarder !!
            }
        } else {
            // Maybe this FutureProxy has been added into FuturePool by readObject
            // Remove it and restore continuation
            ArrayList futures = FuturePool.getIncomingFutures();
            if (futures != null) {
                for (int i = 0; i < futures.size(); i++) {
                    FutureProxy fp = (FutureProxy) futures.get(i);
                    if (fp.creatorID.equals(creatorID) && (fp.ID == ID)) {
                        FuturePool.removeIncomingFutures();
                        continuation = true;
                    }
                }
            }
        }

        // Pass the result
        out.writeObject(target);
        // Pass the continuation flag
        out.writeBoolean(continuation);
        // Pass the id
        out.writeLong(ID);
        //Pass the creatorID
        out.writeObject(creatorID);

        //unset the current continuation tag
        this.continuation = false;
    }

    private synchronized void readObject(java.io.ObjectInputStream in)
        throws java.io.IOException, ClassNotFoundException {
        target = (FutureResult) in.readObject();
        continuation = (boolean) in.readBoolean();
        ID = (long) in.readLong();
        creatorID = (UniqueID) in.readObject();

        // THIS MUST BE DONE WHEN THE SERVER SEND LOGS !!
        if (continuation && isAwaited()) {
            // If we are on a BodyForwarder we DO NOT want to perform the 
            // following operations but can't avoid them.
            // There is a hack in writeObject to restore the continuation state         
            continuation = false;
            FuturePool.registerIncomingFuture(this);
        }

        //now we restore migration to its normal value
        migration = false;
    }

    //
    // -- PRIVATE STATIC METHODS -----------------------------------------------
    //
    private static boolean isFutureObject(Object obj) {
        // If obj is not reified, it cannot be a future
        if (!(MOP.isReifiedObject(obj))) {
            return false;
        }

        // Being a future object is equivalent to have a stub/proxy pair
        // where the proxy object implements the interface FUTURE_PROXY_INTERFACE
        // if the proxy does not inherit from FUTURE_PROXY_ROOT_CLASS
        // it is not a future
        Class proxyclass = ((StubObject) obj).getProxy().getClass();
        Class[] ints = proxyclass.getInterfaces();
        for (int i = 0; i < ints.length; i++) {
            if (Constants.FUTURE_PROXY_INTERFACE.isAssignableFrom(ints[i])) {
                return true;
            }
        }
        return false;
    }

    private static synchronized void setShouldPoolFutureProxyObjects(
        boolean value) {
        if (shouldPoolFutureProxyObjects == value) {
            return;
        }
        shouldPoolFutureProxyObjects = value;
        if (shouldPoolFutureProxyObjects) {
            // Creates the recycle poll for FutureProxy objects
            recyclePool = new FutureProxy[RECYCLE_POOL_SIZE];
            index = 0;
        } else {
            // If we do not want to recycle FutureProxy objects anymore,
            // let's free some memory by permitting the reyclePool to be
            // garbage-collecting
            recyclePool = null;
        }
    }

    private static synchronized void returnFutureProxy(FutureProxy futureProxy) {
        if (!shouldPoolFutureProxyObjects) {
            return;
        }

        // If there's still one slot left in the pool
        if (recyclePool[index] == null) {
            // Cleans up a FutureProxy object
            // It is prefereable to do it here rather than at the moment
            // the object is picked out of the pool, because it allows
            // garbage-collecting the objects referenced in here
            futureProxy.target = null;
            futureProxy.exceptionLevel = null;

            // Inserts the object in the pool
            recyclePool[index] = futureProxy;
            index++;
            if (index == RECYCLE_POOL_SIZE) {
                index = RECYCLE_POOL_SIZE - 1;
            }
        }
    }

    public ExceptionMaskLevel getExceptionLevel() {
        return exceptionLevel;
    }

    public void setExceptionLevel(ExceptionMaskLevel exceptionLevel) {
        this.exceptionLevel = exceptionLevel;
    }

    public synchronized static int futureLength(Object future) {
        int res = 0;
        if ((MOP.isReifiedObject(future)) &&
                ((((StubObject) future).getProxy()) instanceof Future)) {
            res++;
            Future f = (Future) (((StubObject) future).getProxy());
            Object gna = f.getResult();
            while ((MOP.isReifiedObject(gna)) &&
                    ((((StubObject) gna).getProxy()) instanceof Future)) {
                f = (Future) (((StubObject) gna).getProxy());
                gna = f.getResult();
                res++;
            }
        }
        return res;
    }
}

---