Merge pull request #232 from markhamstra/FiniteWait
jobWaiter.synchronized before jobWaiter.wait ...else ``IllegalMonitorStateException`` in ``SimpleFutureAction#ready``.
This commit is contained in:
commit
078049877e
|
@ -99,7 +99,7 @@ class SimpleFutureAction[T] private[spark](jobWaiter: JobWaiter[_], resultFunc:
|
|||
override def ready(atMost: Duration)(implicit permit: CanAwait): SimpleFutureAction.this.type = {
|
||||
if (!atMost.isFinite()) {
|
||||
awaitResult()
|
||||
} else {
|
||||
} else jobWaiter.synchronized {
|
||||
val finishTime = System.currentTimeMillis() + atMost.toMillis
|
||||
while (!isCompleted) {
|
||||
val time = System.currentTimeMillis()
|
||||
|
|
|
@ -31,6 +31,7 @@ private[spark] class JobWaiter[T](
|
|||
private var finishedTasks = 0
|
||||
|
||||
// Is the job as a whole finished (succeeded or failed)?
|
||||
@volatile
|
||||
private var _jobFinished = totalTasks == 0
|
||||
|
||||
def jobFinished = _jobFinished
|
||||
|
|
|
@ -19,6 +19,8 @@ package org.apache.spark.rdd
|
|||
|
||||
import java.util.concurrent.Semaphore
|
||||
|
||||
import scala.concurrent.{Await, TimeoutException}
|
||||
import scala.concurrent.duration.Duration
|
||||
import scala.concurrent.ExecutionContext.Implicits.global
|
||||
|
||||
import org.scalatest.{BeforeAndAfterAll, FunSuite}
|
||||
|
@ -173,4 +175,28 @@ class AsyncRDDActionsSuite extends FunSuite with BeforeAndAfterAll with Timeouts
|
|||
sem.acquire(2)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Awaiting FutureAction results
|
||||
*/
|
||||
test("FutureAction result, infinite wait") {
|
||||
val f = sc.parallelize(1 to 100, 4)
|
||||
.countAsync()
|
||||
assert(Await.result(f, Duration.Inf) === 100)
|
||||
}
|
||||
|
||||
test("FutureAction result, finite wait") {
|
||||
val f = sc.parallelize(1 to 100, 4)
|
||||
.countAsync()
|
||||
assert(Await.result(f, Duration(30, "seconds")) === 100)
|
||||
}
|
||||
|
||||
test("FutureAction result, timeout") {
|
||||
val f = sc.parallelize(1 to 100, 4)
|
||||
.mapPartitions(itr => { Thread.sleep(20); itr })
|
||||
.countAsync()
|
||||
intercept[TimeoutException] {
|
||||
Await.result(f, Duration(20, "milliseconds"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue