Java > Open Source Codes > org > objectweb > proactive > core > body > future > FutureProxy


1 /*
2 * ################################################################
3 *
4 * ProActive: The Java(TM) library for Parallel, Distributed,
5 * Concurrent computing with Security and Mobility
6 *
7 * Copyright (C) 1997-2002 INRIA/University of Nice-Sophia Antipolis
8 * Contact: proactive-support@inria.fr
9 *
10 * This library is free software; you can redistribute it and/or
11 * modify it under the terms of the GNU Lesser General Public
12 * License as published by the Free Software Foundation; either
13 * version 2.1 of the License, or any later version.
14 *
15 * This library is distributed in the hope that it will be useful,
16 * but WITHOUT ANY WARRANTY; without even the implied warranty of
17 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
18 * Lesser General Public License for more details.
19 *
20 * You should have received a copy of the GNU Lesser General Public
21 * License along with this library; if not, write to the Free Software
22 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
23 * USA
24 *
25 * Initial developer(s): The ProActive Team
26 * http://www.inria.fr/oasis/ProActive/contacts.html
27 * Contributor(s):
28 *
29 * ################################################################
30 */
31
32 package org.objectweb.proactive.core.body.future;
33
34 import org.objectweb.proactive.core.Constants;
35 import org.objectweb.proactive.core.ProActiveRuntimeException;
36 import org.objectweb.proactive.core.mop.ConstructionOfReifiedObjectFailedException;
37 import org.objectweb.proactive.core.mop.ConstructorCall;
38 import org.objectweb.proactive.core.mop.MOP;
39 import org.objectweb.proactive.core.mop.MethodCall;
40 import org.objectweb.proactive.core.mop.MethodCallExecutionFailedException;
41 import org.objectweb.proactive.core.mop.Proxy;
42 import org.objectweb.proactive.core.mop.StubObject;
43 import java.lang.reflect.InvocationTargetException ;
44 import org.objectweb.proactive.core.UniqueID;
45 import org.objectweb.proactive.Body;
46 import org.objectweb.proactive.ProActive;
47 import org.objectweb.proactive.core.body.LocalBodyStore;
48 import org.objectweb.proactive.core.body.UniversalBody;
49 import org.objectweb.proactive.core.event.FutureEvent;
50
51
52 /**
53  * This proxy class manages the semantic of future objects
54  *
55  * @author Julien Vayssi?re - INRIA
56  * @see org.objectweb.proactive.core.mop.Proxy
57  *
58  */
59 public class FutureProxy implements Future, Proxy, java.io.Serializable {
60
61     //
62 // -- STATIC MEMBERS -----------------------------------------------
63 //
64
65     /**
66      * The size of the pool we use for recycling FutureProxy objects.
67      */
68     public static final int RECYCLE_POOL_SIZE = 1000;
69     private static FutureProxy[] recyclePool;
70
71     /**
72      * Indicates if the recycling of FutureProxy objects is on.
73      */
74     private static boolean shouldPoolFutureProxyObjects;
75
76     private static int index;
77
78     /** Static point for management of events related to futures.
79      * This FutureEventProducer is responsible for all FutureProxys of this VM */
80     private static FutureEventProducerImpl futureEventProducer;
81
82
83     //
84 // -- PROTECTED MEMBERS -----------------------------------------------
85 //
86
87     /**
88      * The object the proxy sends calls to
89      */
90     protected Object target;
91
92     /**
93      * To mark the Proxy before migration
94      * Usually, the Proxy cannot be serialized if the result is not available (no automatic continuation)
95      * but if we migrate, we don't want to wait for the result
96      */
97     protected boolean migration;
98
99     /**
100      * To mark the proxy before sending this future by parameter or by result
101      */
102     protected boolean continuation;
103
104     /**
105      * UniqueID of the body which create this future
106      */
107     protected UniqueID creatorID;
108
109     /**
110      * ID of the future
111      * In fact, the sequence number of the request that generate this future
112      */
113     protected long ID;
114
115     /**
116      * Unique ID of the sender (in case of automatic continuation).
117      */
118     protected UniqueID senderID;
119
120     /**
121      * This flag indicates the status of the future object
122      */
123     protected boolean isAvailable;
124
125     /**
126      * Indicates if the returned object is an exception
127      */
128     protected boolean isException;
129
130     //
131 // -- CONSTRUCTORS -----------------------------------------------
132 //
133
134     /**
135      * As this proxy does not create a reified object (as opposed to
136      * BodyProxy for example), it is the noargs constructor that
137      * is usually called.
138      */
139
140     public FutureProxy() throws ConstructionOfReifiedObjectFailedException {
141     }
142
143     /**
144      * This constructor is provided for compatibility with other proxies.
145      * More precisely, this permits proxy instanciation via the Meta.newMeta
146      * method.
147      */
148     public FutureProxy(ConstructorCall c, Object [] p) throws ConstructionOfReifiedObjectFailedException {
149         // we don't care what the arguments are
150 this();
151     }
152
153     //
154 // -- PUBLIC STATIC METHODS -----------------------------------------------
155 //
156
157     /**
158      * Tests if the object <code>obj</code> is awaited or not. Always returns
159      * <code>false</code> if <code>obj</code> is not a future object.
160      */
161     public static boolean isAwaited(Object obj) {
162         // If the object is not reified, it cannot be a future
163 if ((MOP.isReifiedObject(obj)) == false)
164             return false;
165         Proxy theProxy = ((StubObject) obj).getProxy();
166         // If it is reified but its proxy is not of type future, we cannot wait
167 if (!(theProxy instanceof Future))
168             return false;
169         return ((Future) theProxy).isAwaited();
170     }
171
172     public synchronized static FutureProxy getFutureProxy() {
173         FutureProxy result;
174         if (shouldPoolFutureProxyObjects && (index > 0)) {
175             // gets the object from the pool
176 index--;
177             result = recyclePool[index];
178             recyclePool[index] = null;
179         } else {
180             try {
181                 result = new FutureProxy();
182             } catch (ConstructionOfReifiedObjectFailedException e) {
183                 result = null;
184             }
185         }
186         return result;
187     }
188
189     /** Returns the <code>FutureEventProducer</code> that is responsible for the
190      * FutureProxys of this VM. Listeners can register themselves here. */
191     public static FutureEventProducer getFutureEventProducer() {
192       if (futureEventProducer == null) futureEventProducer = new FutureEventProducerImpl();
193       return futureEventProducer;
194     }
195
196     //
197 // -- PUBLIC METHODS -----------------------------------------------
198 //
199
200     public boolean equals(Object obj) {
201         //we test if we have a future object
202 if (isFutureObject(obj)) {
203             return (((StubObject) obj).getProxy().hashCode() == this.hashCode());
204         }
205         return false;
206     }
207
208     //
209 // -- Implements Future -----------------------------------------------
210 //
211
212     /**
213      * Invoked by a thread of the skeleton that performed the service in order
214      * to tie the result object to the proxy.
215      *
216      * If the execution of the call raised an exception, this exception is put
217      * into an object of class InvocationTargetException and returned, just like
218      * for any returned object
219      */
220
221     public synchronized void receiveReply(Object obj) throws java.io.IOException {
222         if (target != null) {
223             throw new java.io.IOException ("FutureProxy receives a reply and this target field is not null");
224         }
225         target = obj;
226         if (target != null) {
227             isException = (target instanceof Throwable );
228         }
229         isAvailable = true;
230         this.notifyAll();
231     }
232
233     /**
234      * Returns the result this future is for as an exception if an exception has been raised
235      * or null if the result is not an exception. The method blocks until the result is available.
236      * @return the exception raised once available or null if no exception.
237      */
238     public synchronized Throwable getRaisedException() {
239         waitFor();
240         if (isException)
241             return (Throwable ) target;
242         return null;
243     }
244
245     /**
246      * Returns the result this future is for. The method blocks until the future is available
247      * @return the result of this future object once available.
248      */
249     public synchronized Object getResult() {
250         waitFor();
251         return target;
252     }
253
254
255     public synchronized void setResult(Object o) {
256         target = o;
257         isAvailable = true;
258     }
259
260     /**
261      * Tests the status of the returned object
262      * @return <code>true</code> if the future object is NOT yet available, <code>false</code> if it is.
263      */
264     public synchronized boolean isAwaited() {
265         return !isAvailable;
266     }
267
268     /**
269      * Blocks the calling thread until the future object is available.
270      */
271     public synchronized void waitFor() {
272
273         if (isAvailable) return;
274         
275         UniqueID id = null;
276         
277         // send WAIT_BY_NECESSITY event to listeners if there are any
278 if (futureEventProducer != null) {
279           id = ProActive.getBodyOnThis().getID();
280           if (LocalBodyStore.getInstance().getLocalBody(id) != null) {
281             // send event only if ActiveObject, not for HalfBodies
282 futureEventProducer.notifyListeners(id,
283                                               getCreatorID(), FutureEvent.WAIT_BY_NECESSITY);
284           } else id = null;
285         }
286         while (!isAvailable) {
287             try {
288                 this.wait();
289             } catch (InterruptedException e) {
290             }
291         }
292         // send RECEIVED_FUTURE_RESULT event to listeners if there are any
293 if (id != null) {
294           futureEventProducer.notifyListeners(id,
295                                               getCreatorID(), FutureEvent.RECEIVED_FUTURE_RESULT);
296         }
297
298     }
299     
300     
301     
302     public long getID() {
303         return ID;
304     }
305
306     public void setID(long l) {
307         ID = l;
308     }
309
310     public void setCreatorID(UniqueID i) {
311         creatorID = i;
312     }
313
314     public UniqueID getCreatorID() {
315         return creatorID;
316     }
317
318     public void setSenderID(UniqueID i) {
319         senderID = i;
320     }
321
322
323
324     //
325 // -- Implements Proxy -----------------------------------------------
326 //
327
328     /**
329      * Blocks until the future object is available, then executes Call <code>c</code> on the now-available object.
330      *
331      * As future and process behaviors are mutually exclusive, we know that
332      * the invocation of a method on a future objects cannot lead to wait-by
333      * necessity. Thus, we can propagate all exceptions raised by this invocation
334      *
335      * @exception InvocationTargetException If the invokation of the method represented by the
336      * <code>Call</code> object <code>c</code> on the reified object
337      * throws an exception, this exception is thrown as-is here. The stub then
338      * throws this exception to the calling thread after checking that it is
339      * declared in the throws clause of the reified method. Otherwise, the stub
340      * does nothing except print a message on System.err (or out ?).
341      */
342     public Object reify(MethodCall c) throws InvocationTargetException {
343         Object result = null;
344         //stem.out.println("FutureProxy: c.getName() = " +c.getName());
345 // if ((c.getName()).equals("equals") || (c.getName()).equals("hashCode")) {
346 // //System.out.println("FutureProxy: now executing " + c.getName());
347 // try {
348 // result = c.execute(this);
349 // } catch (MethodCallExecutionFailedException e) {
350 // throw new ProActiveRuntimeException("FutureProxy: Illegal arguments in call " + c.getName());
351 // }
352 // return result;
353 // }
354 waitFor();
355     
356         // Now that the object is available, execute the call
357 if (this.isException) {
358             throw ((InvocationTargetException ) this.target);
359         } else {
360             try {
361                 result = c.execute(this.target);
362             } catch (MethodCallExecutionFailedException e) {
363                 throw new ProActiveRuntimeException("FutureProxy: Illegal arguments in call " + c.getName());
364             }
365         }
366         
367         // If target of this future is another future, make a shortcut !
368 if (target instanceof StubObject) {
369             Proxy p =((StubObject)target).getProxy();
370             if ( p instanceof FutureProxy ) {
371                 target = ((FutureProxy)p).target;
372             }
373         }
374         
375         return result;
376     }
377
378     //
379 // -- PROTECTED METHODS -----------------------------------------------
380 //
381
382     protected void finalize() {
383         returnFutureProxy(this);
384     }
385
386     protected void setMigrationTag() {
387         migration = true;
388     }
389
390     protected void unsetMigrationTag() {
391         migration = false;
392     }
393     
394     public synchronized void setContinuationTag(){
395         continuation = true;
396     }
397     
398     public synchronized void unsetContinuationTag(){
399         continuation = false;
400     }
401     
402     
403
404     
405     //
406 // -- PRIVATE METHODS FOR SERIALIZATION -----------------------------------------------
407 //
408
409     private synchronized void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
410         
411         
412         //if continuation is already set, we are in a forwarder
413 //else if a destination is available in destTable, set the continuation tag
414 if (!continuation) {
415             continuation = (FuturePool.getBodyDestination()!=null);
416         }
417         
418         // We wait until the result is available
419 if ((!migration) && (!continuation)) {
420             waitFor();
421         }
422
423         // Registration in case of continuation
424 if (continuation && isAwaited()) {
425             // get the sender body
426 Body sender = LocalBodyStore.getInstance().getLocalBody(senderID);
427             // it's a halfbody...
428 if (sender == null) {
429                 sender = LocalBodyStore.getInstance().getLocalHalfBody(senderID);
430             }
431             if (sender != null) {
432                 UniversalBody dest = FuturePool.getBodyDestination();
433                 if (dest != null) {
434                     sender.getFuturePool().addAutomaticContinuation(ID, creatorID, dest);
435                 }
436             }// if sender is still null, it's a forwarder !!
437 }
438
439         // Pass the result
440 out.writeObject(target);
441         // Pass the continuation flag
442 out.writeBoolean(continuation);
443         // Pass the id
444 out.writeLong(ID);
445         //Pass the creatorID
446 out.writeObject(creatorID);
447
448         // It is impossible that a future object can be passed
449 // as a parameter if it has raised a checked exception
450 // For the other exceptions...
451 out.writeBoolean(isException);
452         out.writeBoolean(isAvailable);
453         
454         //unset the current continuation tag
455 this.continuation = false;
456     }
457
458
459     private synchronized void readObject(java.io.ObjectInputStream in) throws java.io.IOException , ClassNotFoundException {
460         target = (Object ) in.readObject();
461         continuation = (boolean) in.readBoolean();
462         ID = (long) in.readLong();
463         creatorID = (UniqueID) in.readObject();
464         isException = (boolean) in.readBoolean();
465         isAvailable = (boolean) in.readBoolean();
466
467         if (continuation && isAwaited()) {
468             continuation = false;
469             FuturePool.registerIncomingFuture(this);
470                 
471         }
472
473         //now we restore migration to its normal value
474 migration = false;
475     }
476
477     //
478 // -- PRIVATE STATIC METHODS -----------------------------------------------
479 //
480
481     private static boolean isFutureObject(Object obj) {
482         // If obj is not reified, it cannot be a future
483 if (!(MOP.isReifiedObject(obj)))
484             return false;
485         // Being a future object is equivalent to have a stub/proxy pair
486 // where the proxy object implements the interface FUTURE_PROXY_INTERFACE
487 // if the proxy does not inherit from FUTURE_PROXY_ROOT_CLASS
488 // it is not a future
489 Class proxyclass = ((StubObject) obj).getProxy().getClass();
490         Class [] ints = proxyclass.getInterfaces();
491         for (int i = 0; i < ints.length; i++) {
492             if (Constants.FUTURE_PROXY_INTERFACE.isAssignableFrom(ints[i]))
493                 return true;
494         }
495         return false;
496     }
497
498     private static synchronized void setShouldPoolFutureProxyObjects(boolean value) {
499         if (shouldPoolFutureProxyObjects == value)
500             return;
501         shouldPoolFutureProxyObjects = value;
502         if (shouldPoolFutureProxyObjects) {
503             // Creates the recycle poll for FutureProxy objects
504 recyclePool = new FutureProxy[RECYCLE_POOL_SIZE];
505             index = 0;
506         } else {
507             // If we do not want to recycle FutureProxy objects anymore,
508 // let's free some memory by permitting the reyclePool to be
509 // garbage-collecting
510 recyclePool = null;
511         }
512     }
513
514     private static synchronized void returnFutureProxy(FutureProxy futureProxy) {
515         if (!shouldPoolFutureProxyObjects)
516             return;
517         // If there's still one slot left in the pool
518 if (recyclePool[index] == null) {
519             // Cleans up a FutureProxy object
520 // It is prefereable to do it here rather than at the moment
521 // the object is picked out of the pool, because it allows
522 // garbage-collecting the objects referenced in here
523 futureProxy.target = null;
524             futureProxy.isAvailable = false;
525             futureProxy.isException = false;
526
527             // Inserts the object in the pool
528 recyclePool[index] = futureProxy;
529             index++;
530             if (index == RECYCLE_POOL_SIZE)
531                 index = RECYCLE_POOL_SIZE - 1;
532         }
533     }
534
535
536
537
538     //////////////////////////
539 //////////////////////////
540 ////FOR DEBUG PURPOSE/////
541 //////////////////////////
542 //////////////////////////
543 public synchronized static int futureLength(Object future) {
544         int res = 0;
545         if ((MOP.isReifiedObject(future))
546             && ((((StubObject) future).getProxy()) instanceof Future)) {
547             res++;
548             Future f = (Future) (((StubObject) future).getProxy());
549             Object gna = f.getResult();
550             while ((MOP.isReifiedObject(gna))
551                 && ((((StubObject) gna).getProxy()) instanceof Future)) {
552                 f = (Future) (((StubObject) gna).getProxy());
553                 gna = f.getResult();
554                 res++;
555             }
556         }
557         return res;
558     }
559     
560     
561 }
562

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags