8
8
9
9
package scala .concurrent
10
10
11
- import java .util .concurrent .{ Executor }
12
- import scala .concurrent ._
11
+ import java .util .concurrent .Executor
13
12
import scala .annotation .tailrec
14
13
15
- /**
16
- * All Batchables are automatically batched when submitted to a BatchingExecutor
17
- */
18
- private [concurrent] trait Batchable extends Runnable {
19
- def isBatchable : Boolean
20
- }
21
-
22
14
/**
23
15
* Mixin trait for an Executor
24
16
* which groups multiple nested `Runnable.run()` calls
@@ -64,13 +56,13 @@ private[concurrent] trait BatchingExecutor extends Executor {
64
56
parentBlockContext = prevBlockContext
65
57
66
58
@ tailrec def processBatch (batch : List [Runnable ]): Unit = batch match {
67
- case Nil ⇒ ()
68
- case head :: tail ⇒
59
+ case Nil => ()
60
+ case head :: tail =>
69
61
_tasksLocal set tail
70
62
try {
71
63
head.run()
72
64
} catch {
73
- case t : Throwable ⇒
65
+ case t : Throwable =>
74
66
// if one task throws, move the
75
67
// remaining tasks to another thread
76
68
// so we can throw the exception
@@ -91,7 +83,7 @@ private[concurrent] trait BatchingExecutor extends Executor {
91
83
}
92
84
}
93
85
94
- override def blockOn [T ](thunk : ⇒ T )(implicit permission : CanAwait ): T = {
86
+ override def blockOn [T ](thunk : => T )(implicit permission : CanAwait ): T = {
95
87
// if we know there will be blocking, we don't want to keep tasks queued up because it could deadlock.
96
88
{
97
89
val tasks = _tasksLocal.get
@@ -111,16 +103,15 @@ private[concurrent] trait BatchingExecutor extends Executor {
111
103
override def execute (runnable : Runnable ): Unit = {
112
104
if (batchable(runnable)) { // If we can batch the runnable
113
105
_tasksLocal.get match {
114
- case null ⇒ unbatchedExecute(new Batch (List (runnable))) // If we aren't in batching mode yet, enqueue batch
115
- case some ⇒ _tasksLocal.set(runnable :: some) // If we are already in batching mode, add to batch
106
+ case null => unbatchedExecute(new Batch (List (runnable))) // If we aren't in batching mode yet, enqueue batch
107
+ case some => _tasksLocal.set(runnable :: some) // If we are already in batching mode, add to batch
116
108
}
117
109
} else unbatchedExecute(runnable) // If not batchable, just delegate to underlying
118
110
}
119
111
120
112
/** Override this to define which runnables will be batched. */
121
113
def batchable (runnable : Runnable ): Boolean = runnable match {
122
- case b : Batchable ⇒ b.isBatchable
123
- case _ : scala.concurrent.OnCompleteRunnable ⇒ true
124
- case _ ⇒ false
114
+ case _ : OnCompleteRunnable => true
115
+ case _ => false
125
116
}
126
117
}
0 commit comments