1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
/* __ *\
** ________ ___ / / ___ Scala API **
** / __/ __// _ | / / / _ | (c) 2003-2011, LAMP/EPFL **
** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ **
** /____/\___/_/ |_/____/_/ | | **
** |/ **
\* */
package scala
import scala.util.{ Timeout, Duration }
/** This package object contains primitives for concurrent and parallel programming.
*/
package object concurrent {
type ExecutionException = java.util.concurrent.ExecutionException
type CancellationException = java.util.concurrent.CancellationException
type TimeoutException = java.util.concurrent.TimeoutException
/** A global execution environment for executing lightweight tasks.
*/
lazy val executionContext =
new default.ExecutionContextImpl
/** A global service for scheduling tasks for execution.
*/
lazy val scheduler =
new default.SchedulerImpl
private[concurrent] def currentExecutionContext: ThreadLocal[ExecutionContext] = new ThreadLocal[ExecutionContext] {
override protected def initialValue = null
}
val handledFutureException: PartialFunction[Throwable, Throwable] = {
case t: Throwable if isFutureThrowable(t) => t
}
// TODO rename appropriately and make public
private[concurrent] def isFutureThrowable(t: Throwable) = t match {
case e: Error => false
case t: scala.util.control.ControlThrowable => false
case i: InterruptedException => false
case _ => true
}
private[concurrent] def resolve[T](source: Either[Throwable, T]): Either[Throwable, T] = source match {
case Left(t: scala.runtime.NonLocalReturnControl[_]) => Right(t.value.asInstanceOf[T])
case Left(t: InterruptedException) => Left(new ExecutionException("Boxed InterruptedException", t))
case Left(e: Error) => throw e
case _ => source
}
private val resolverFunction: PartialFunction[Throwable, Either[Throwable, _]] = {
case t: scala.runtime.NonLocalReturnControl[_] => Right(t.value)
case t: InterruptedException => Left(new ExecutionException("Boxed InterruptedException", t))
case e: Error => throw e
case t => Left(t)
}
private[concurrent] def resolver[T] = resolverFunction.asInstanceOf[PartialFunction[Throwable, Either[Throwable, T]]]
/* concurrency constructs */
def future[T](body: =>T)(implicit execCtx: ExecutionContext = executionContext): Future[T] =
execCtx future body
def promise[T]()(implicit execCtx: ExecutionContext = executionContext): Promise[T] =
execCtx promise
/** Used to block on a piece of code which potentially blocks.
*
* @param body A piece of code which contains potentially blocking or long running calls.
*
* Calling this method may throw the following exceptions:
* - CancellationException - if the computation was cancelled
* - InterruptedException - in the case that a wait within the blockable object was interrupted
* - TimeoutException - in the case that the blockable object timed out
*/
def await[T](atMost: Duration)(body: =>T): T = result(new Awaitable[T] {
def await(timeout: Timeout)(implicit cb: CanAwait) = body
}, atMost)
/** Blocks on a blockable object.
*
* @param awaitable An object with a `block` method which runs potentially blocking or long running calls.
*
* Calling this method may throw the following exceptions:
* - CancellationException - if the computation was cancelled
* - InterruptedException - in the case that a wait within the blockable object was interrupted
* - TimeoutException - in the case that the blockable object timed out
*/
def result[T](awaitable: Awaitable[T], atMost: Duration): T = {
currentExecutionContext.get match {
case null => awaitable.await(atMost)(null) // outside - TODO - fix timeout case
case x => x.blockingCall(atMost, awaitable) // inside an execution context thread
}
}
def ready[T](awaitable: Awaitable[T], atMost: Duration): Awaitable[T] = {
result(awaitable, atMost)
awaitable
}
}
package concurrent {
/** A timeout exception.
*
* Futures are failed with a timeout exception when their timeout expires.
*
* Each timeout exception contains an origin future which originally timed out.
*/
class FutureTimeoutException(origin: Future[_], message: String) extends TimeoutException(message) {
def this(origin: Future[_]) = this(origin, "Future timed out.")
}
}
|
