[Swift-commit] cog r4024

[swift at ci.uchicago.edu]

------------------------------------------------------------------------
r4024 | hategan | 2014-07-13 02:34:47 -0500 (Sun, 13 Jul 2014) | 1 line

fixing future notification loops: each relevant yield gets a sequence number that represents the state of the (non-binary) future; addListener must receive the yield as a parameter; if the future sequence is the same as the yield sequence, it means that the future has not changed in the mean time, so no notification is necessary
------------------------------------------------------------------------
Index: modules/karajan/src/org/globus/cog/karajan/futures/ChannelSplitter.java
===================================================================
--- modules/karajan/src/org/globus/cog/karajan/futures/ChannelSplitter.java	(revision 4023)
+++ modules/karajan/src/org/globus/cog/karajan/futures/ChannelSplitter.java	(working copy)
@@ -32,7 +32,7 @@
 		for (int i = 0; i < count; i++) {
 			out[i] = new FutureMemoryChannel<T>();
 		}
-		c.addListener(this);
+		c.addListener(this, null);
 	}
 	
 	public FutureMemoryChannel<T>[] getOuts() {
Index: modules/karajan/src/k/rt/ConditionalYield.java
===================================================================
--- modules/karajan/src/k/rt/ConditionalYield.java	(revision 4023)
+++ modules/karajan/src/k/rt/ConditionalYield.java	(working copy)
@@ -13,17 +13,29 @@
 
 public class ConditionalYield extends Yield {
     private Future f;
+    private final int sequence;
 
     public ConditionalYield(Future f) {
         this.f = f;
+        this.sequence = -1;
     }
     
+    public ConditionalYield(Future f, int sequence) {
+        this.f = f;
+        this.sequence = sequence;
+    }
+    
     public ConditionalYield(int pstate, Future f) {
     	super(pstate);
         this.f = f;
+        this.sequence = -1;
     }
 
     public Future getFuture() {
         return f;
     }
+
+	public int getSequence() {
+		return sequence;
+	}
 }
Index: modules/karajan/src/k/rt/Future.java
===================================================================
--- modules/karajan/src/k/rt/Future.java	(revision 4023)
+++ modules/karajan/src/k/rt/Future.java	(working copy)
@@ -10,5 +10,5 @@
 package k.rt;
 
 public interface Future {
-    void addListener(FutureListener l);
+    void addListener(FutureListener l, ConditionalYield y);
 }
Index: modules/karajan/src/k/rt/FutureMemoryChannel.java
===================================================================
--- modules/karajan/src/k/rt/FutureMemoryChannel.java	(revision 4023)
+++ modules/karajan/src/k/rt/FutureMemoryChannel.java	(working copy)
@@ -57,7 +57,7 @@
             return super.get(index);
         }
         else {
-            throw new ConditionalYield(this);
+            throw new ConditionalYield(this, super.size());
         }
     }
     
@@ -80,7 +80,7 @@
             return super.getAll();
         }
         else {
-            throw new ConditionalYield(this);
+            throw new ConditionalYield(this, super.size());
         }
     }
 
@@ -106,17 +106,17 @@
             return super.size();
         }
         else {
-            throw new ConditionalYield(this);
+            throw new ConditionalYield(this, super.size());
         }
     }
 
     @Override
-    public synchronized void addListener(FutureListener l) {
+    public synchronized void addListener(FutureListener l, ConditionalYield y) {
         if (listeners == null) {
             listeners = new LinkedList<FutureListener>();
         }
         listeners.add(l);
-        if (closed) {
+        if (closed || (y != null && (y.getSequence() != super.size()))) {
             notifyListeners();
         }
     }
Index: modules/karajan/src/k/rt/AbstractFuture.java
===================================================================
--- modules/karajan/src/k/rt/AbstractFuture.java	(revision 4023)
+++ modules/karajan/src/k/rt/AbstractFuture.java	(working copy)
@@ -18,7 +18,7 @@
 	protected abstract boolean isClosed();
 	
 	@Override
-	public void addListener(FutureListener l) {
+	public void addListener(FutureListener l, ConditionalYield y) {
 		boolean closed;
 		synchronized(this) {
 			if (listeners == null) {
Index: modules/karajan/src/k/thr/ThreadSet.java
===================================================================
--- modules/karajan/src/k/thr/ThreadSet.java	(revision 4023)
+++ modules/karajan/src/k/thr/ThreadSet.java	(working copy)
@@ -100,7 +100,7 @@
 	}
 
 	@Override
-	public synchronized void addListener(FutureListener l) {
+	public synchronized void addListener(FutureListener l, ConditionalYield y) {
 		if (listener != null) {
 			throw new IllegalThreadStateException("Multiple listeners");
 		}
Index: modules/karajan/src/k/thr/ThreadSetFixed.java
===================================================================
--- modules/karajan/src/k/thr/ThreadSetFixed.java	(revision 4023)
+++ modules/karajan/src/k/thr/ThreadSetFixed.java	(working copy)
@@ -95,7 +95,7 @@
 	}
 
 	@Override
-	public synchronized void addListener(FutureListener l) {
+	public synchronized void addListener(FutureListener l, ConditionalYield y) {
 		if (listener != null) {
 			throw new IllegalThreadStateException("Multiple listeners");
 		}
Index: modules/karajan/src/k/thr/LWThread.java
===================================================================
--- modules/karajan/src/k/thr/LWThread.java	(revision 4023)
+++ modules/karajan/src/k/thr/LWThread.java	(working copy)
@@ -247,7 +247,7 @@
             	sleeping = setSleeping();
             }
             if (sleeping) {
-            	e.getFuture().addListener(new Listener());
+            	e.getFuture().addListener(new Listener(), e);
             }
         }
         catch (WaitYield e) {
Index: modules/karajan/src/k/thr/ThrottledThreadSet.java
===================================================================
--- modules/karajan/src/k/thr/ThrottledThreadSet.java	(revision 4023)
+++ modules/karajan/src/k/thr/ThrottledThreadSet.java	(working copy)
@@ -26,13 +26,13 @@
 		}
 
 		@Override
-		public void addListener(FutureListener l) {
+		public void addListener(FutureListener l, ConditionalYield y) {
 		    synchronized(ThrottledThreadSet.this) {
     		    if (canAdd()) {
     		        l.futureUpdated(this);
     		    }
     		    else {
-    		    	super.addListener(l);
+    		    	super.addListener(l, y);
     		    }
 		    }
 		}