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.concurrent.util.Duration
import scala.util.{ Try, Success, Failure }
import scala.concurrent.forkjoin.{ ForkJoinPool, RecursiveTask => FJTask, RecursiveAction, ForkJoinWorkerThread }
import scala.collection.generic.CanBuildFrom
import collection._
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
def reportFailure(t: Throwable): Unit
/* 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 Failure(t) =>
p tryFailure t
case Success(v) =>
buffer(currentIndex) = v
tryFinish()
}
idx += 1
}
tryFinish()
p.future
}
/** TODO some docs
*
*/
def any[T](futures: Traversable[Future[T]]): Future[T] = {
val p = promise[T]
val completeFirst: Try[T] => Unit = elem => p tryComplete elem
futures foreach (_ onComplete completeFirst)
p.future
}
/** TODO some docs
*
*/
def find[T](futures: Traversable[Future[T]])(predicate: T => Boolean): 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: Try[T] => Unit = {
v => v match {
case Success(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
|
