SyncVar enhancements:

* deprecation of "exception" field
* addition of blocking-queue-like "take", "put" methods
* change of "unset" signature

4 files changed, 95 insertions, 7 deletions

diff --git a/src/library/scala/concurrent/SyncVar.scala b/src/library/scala/concurrent/SyncVar.scala
index 6b9ca33395..c04268990a 100644
--- a/src/library/scala/concurrent/SyncVar.scala
+++ b/src/library/scala/concurrent/SyncVar.scala
@@ -28,6 +28,14 @@ class SyncVar[A] {
     else throw exception.get
   }
 
+  def take() = synchronized {
+    try {
+      get
+    } finally {
+      unset()
+    }
+  }
+
   def set(x: A) = synchronized {
     value = x
     isDefined = true
@@ -41,6 +49,9 @@ class SyncVar[A] {
     notifyAll()
   }
 
+  /**
+   *  @deprecated Will be removed in 2.8. SyncVar should not allow exception by design.
+   */
   def setWithCatch(x: => A) = synchronized {
     try {
       this set x
@@ -51,12 +62,18 @@ class SyncVar[A] {
     }
   }
 
+  def put(x: A) = synchronized {
+    while (isDefined) wait()
+    set(x)
+  }
+
   def isSet: Boolean = synchronized {
     isDefined
   }
 
-  def unset = synchronized {
+  def unset(): Unit = synchronized {
     isDefined = false
+    notifyAll()
   }
 
 }
diff --git a/src/library/scala/concurrent/ops.scala b/src/library/scala/concurrent/ops.scala
index 0d3c485300..3bb0fb9857 100644
--- a/src/library/scala/concurrent/ops.scala
+++ b/src/library/scala/concurrent/ops.scala
@@ -20,6 +20,17 @@ import java.lang.Thread
  *  @version 1.0, 12/03/2003
  */
 object ops {
+  /**
+   *  If expression computed successfully return it in <code>Left</code>,
+   *  otherwise return exception in <code>Right</code>.
+   */
+  private def tryCatch[A](left: => A): Either[A, Throwable] = {
+    try {
+      Left(left)
+    } catch {
+      case t => Right(t)
+    }
+  }
 
   /**
    *  @param p ...
@@ -34,9 +45,12 @@ object ops {
    *  @return  ...
    */
   def future[A](p: => A): () => A = {
-    val result = new SyncVar[A]
-    spawn { result setWithCatch p }
-    () => result.get
+    val result = new SyncVar[Either[A, Throwable]]
+    spawn { result set tryCatch(p) }
+    () => result.get match {
+    	case Left(a) => a
+    	case Right(t) => throw t
+    }
   }
 
   /**
@@ -45,9 +59,12 @@ object ops {
    *  @return   ...
    */
   def par[A, B](xp: => A, yp: => B): (A, B) = {
-    val y = new SyncVar[B]
-    spawn { y setWithCatch yp }
-    (xp, y.get)
+    val y = new SyncVar[Either[B, Throwable]]
+    spawn { y set tryCatch(yp) }
+    (xp, y.get match {
+    	case Left(b) => b
+    	case Right(t) => throw t
+    })
   }
 
   /**
diff --git a/test/files/run/sync-var.check b/test/files/run/sync-var.check
new file mode 100644
index 0000000000..e77aa319a5
--- /dev/null
+++ b/test/files/run/sync-var.check
@@ -0,0 +1 @@
+50005000 50005000 true
diff --git a/test/files/run/sync-var.scala b/test/files/run/sync-var.scala
new file mode 100644
index 0000000000..aa6ae9fa34
--- /dev/null
+++ b/test/files/run/sync-var.scala
@@ -0,0 +1,53 @@
+import java.util.concurrent._
+import java.util.concurrent.atomic._
+
+object Test { def main(args: Array[String]) {
+
+val n = 10000
+val i = new AtomicInteger(n)
+val j = new AtomicInteger(n)
+val sum = new AtomicInteger
+
+val q = new scala.concurrent.SyncVar[Int]
+
+val producers = (1 to 3).force map { z => new Thread {
+  override def run() {
+    var again = true
+    while (again) {
+      val x = i.getAndDecrement()
+      if (x > 0)
+        q put x
+      else
+        again = false
+    }
+  }
+} }
+
+val summers = (1 to 7).force map { z => new Thread {
+  override def run() {
+    val x = j.decrementAndGet()
+    if (x >= 0) {
+      sum addAndGet q.take()
+    }
+    if (x > 0) {
+      run()
+    } else {
+      // done
+    }
+  }
+} }
+
+summers foreach { _.start() }
+producers foreach { _.start() }
+
+summers foreach { _.join() }
+
+val got = sum.get
+val expected = (n + 1) * n / 2
+println(got + " " + expected + " " + (got == expected))
+
+producers foreach { _.join() }
+
+} }
+
+// vim: set ts=2 sw=2 et: