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
130
131
132
|
/* __ *\
** ________ ___ / / ___ Scala API **
** / __/ __// _ | / / / _ | (c) 2003-2011, LAMP/EPFL **
** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ **
** /____/\___/_/ |_/____/_/ | | **
** |/ **
\* */
package scala.concurrent
import java.util.concurrent.atomic.{ AtomicInteger }
import java.util.concurrent.{ Executors, Future => JFuture, Callable }
import scala.util.Duration
import scala.concurrent.forkjoin.{ ForkJoinPool, RecursiveTask => FJTask, RecursiveAction, ForkJoinWorkerThread }
import scala.collection.generic.CanBuildFrom
import collection._
import annotation.implicitNotFound
trait ExecutionContext {
protected implicit object CanAwaitEvidence extends CanAwait
def execute(runnable: Runnable): Unit
def execute[U](body: () => U): Unit
def promise[T]: Promise[T]
def future[T](body: Callable[T]): Future[T] = future(body.call())
def future[T](body: => T): Future[T]
def blocking[T](atMost: Duration)(body: =>T): T
def blocking[T](awaitable: Awaitable[T], atMost: Duration): T
/* implementations follow */
private implicit val executionContext = this
def keptPromise[T](result: T): Promise[T] = {
val p = promise[T]
p success result
}
def brokenPromise[T](t: Throwable): Promise[T] = {
val p = promise[T]
p failure t
}
/** TODO some docs
*
*/
def all[T, Coll[X] <: Traversable[X]](futures: Coll[Future[T]])(implicit cbf: CanBuildFrom[Coll[_], T, Coll[T]]): Future[Coll[T]] = {
import nondeterministic._
val buffer = new mutable.ArrayBuffer[T]
val counter = new AtomicInteger(1) // how else could we do this?
val p: Promise[Coll[T]] = promise[Coll[T]] // we need an implicit execctx in the signature
var idx = 0
def tryFinish() = if (counter.decrementAndGet() == 0) {
val builder = cbf(futures)
builder ++= buffer
p success builder.result
}
for (f <- futures) {
val currentIndex = idx
buffer += null.asInstanceOf[T]
counter.incrementAndGet()
f onComplete {
case Left(t) =>
p tryFailure t
case Right(v) =>
buffer(currentIndex) = v
tryFinish()
}
idx += 1
}
tryFinish()
p.future
}
/** TODO some docs
*
*/
@implicitNotFound(msg = "Calling this method yields non-deterministic programs.")
def any[T](futures: Traversable[Future[T]])(implicit nondet: NonDeterministic): Future[T] = {
val p = promise[T]
val completeFirst: Either[Throwable, T] => Unit = elem => p tryComplete elem
futures foreach (_ onComplete completeFirst)
p.future
}
/** TODO some docs
*
*/
@implicitNotFound(msg = "Calling this method yields non-deterministic programs.")
def find[T](futures: Traversable[Future[T]])(predicate: T => Boolean)(implicit nondet: NonDeterministic): Future[Option[T]] = {
if (futures.isEmpty) Promise.kept[Option[T]](None).future
else {
val result = promise[Option[T]]
val count = new AtomicInteger(futures.size)
val search: Either[Throwable, T] => Unit = {
v => v match {
case Right(r) => if (predicate(r)) result trySuccess Some(r)
case _ =>
}
if (count.decrementAndGet() == 0) result trySuccess None
}
futures.foreach(_ onComplete search)
result.future
}
}
}
sealed trait CanAwait
|
