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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
package dotty
package tools
package vulpix
import java.io.{ File => JFile, InputStreamReader, BufferedReader, PrintStream }
import java.util.concurrent.TimeoutException
import scala.concurrent.duration.Duration
import scala.concurrent.{ Await, Future }
import scala.concurrent.ExecutionContext.Implicits.global
import scala.collection.mutable
/** Vulpix spawns JVM subprocesses (`numberOfSlaves`) in order to run tests
* without compromising the main JVM
*
* These need to be orchestrated in a safe manner with a simple protocol. This
* interface provides just that.
*
* The protocol is defined as:
*
* - master sends classpath to for which to run `Test#main` and waits for
* `maxDuration`
* - slave invokes the method and waits until completion
* - upon completion it sends back a `RunComplete` message
* - the master checks if the child is still alive
* - child is still alive, the output was valid
* - child is dead, the output is the failure message
*
* If this whole chain of events is not completed within `maxDuration`, the
* child process is destroyed and a new child is spawned.
*/
trait RunnerOrchestration {
/** The maximum amount of active runners, which contain a child JVM */
def numberOfSlaves: Int
/** The maximum duration the child process is allowed to consume before
* getting destroyed
*/
def maxDuration: Duration
/** Destroy and respawn process after each test */
def safeMode: Boolean
/** Running a `Test` class's main method from the specified `dir` */
def runMain(classPath: String): Status = monitor.runMain(classPath)
private[this] val monitor = new RunnerMonitor
/** Look away now, sweet child of summer */
private class RunnerMonitor {
def runMain(classPath: String): Status = withRunner(_.runMain(classPath))
private class Runner(private var process: Process) {
private[this] var childStdout: BufferedReader = _
private[this] var childStdin: PrintStream = _
/** Checks if `process` is still alive
*
* When `process.exitValue()` is called on an active process the caught
* exception is thrown. As such we can know if the subprocess exited or
* not.
*/
def isAlive: Boolean =
try { process.exitValue(); false }
catch { case _: IllegalThreadStateException => true }
/** Destroys the underlying process and kills IO streams */
def kill(): Unit = {
if (process ne null) process.destroy()
process = null
childStdout = null
childStdin = null
}
/** Blocks less than `maxDuration` while running `Test.main` from `dir` */
def runMain(classPath: String): Status = {
assert(process ne null,
"Runner was killed and then reused without setting a new process")
// Makes the encapsulating RunnerMonitor spawn a new runner
def respawn(): Unit = {
process.destroy()
process = createProcess
childStdout = null
childStdin = null
}
if (childStdin eq null)
childStdin = new PrintStream(process.getOutputStream, /* autoFlush = */ true)
// pass file to running process
childStdin.println(classPath)
// Create a future reading the object:
val readOutput = Future {
val sb = new StringBuilder
if (childStdout eq null)
childStdout = new BufferedReader(new InputStreamReader(process.getInputStream))
var childOutput = childStdout.readLine()
while (childOutput != ChildJVMMain.MessageEnd && childOutput != null) {
sb.append(childOutput)
sb += '\n'
childOutput = childStdout.readLine()
}
if (process.isAlive && childOutput != null) Success(sb.toString)
else Failure(sb.toString)
}
// Await result for `maxDuration` and then timout and destroy the
// process:
val status =
try Await.result(readOutput, maxDuration)
catch { case _: TimeoutException => Timeout }
// Handle failure of the VM:
status match {
case _: Success if safeMode => respawn()
case _: Failure => respawn()
case Timeout => respawn()
case _ => ()
}
status
}
}
private def createProcess: Process = {
val sep = sys.props("file.separator")
val cp =
classOf[ChildJVMMain].getProtectionDomain.getCodeSource.getLocation.getFile + ":" +
Jars.scalaLibraryFromRuntime
val javaBin = sys.props("java.home") + sep + "bin" + sep + "java"
new ProcessBuilder(javaBin, "-cp", cp, "dotty.tools.vulpix.ChildJVMMain")
.redirectErrorStream(true)
.redirectInput(ProcessBuilder.Redirect.PIPE)
.redirectOutput(ProcessBuilder.Redirect.PIPE)
.start()
}
private[this] val allRunners = List.fill(numberOfSlaves)(new Runner(createProcess))
private[this] val freeRunners = mutable.Queue(allRunners: _*)
private[this] val busyRunners = mutable.Set.empty[Runner]
private def getRunner(): Runner = synchronized {
while (freeRunners.isEmpty) wait()
val runner = freeRunners.dequeue()
busyRunners += runner
notify()
runner
}
private def freeRunner(runner: Runner): Unit = synchronized {
freeRunners.enqueue(runner)
busyRunners -= runner
notify()
}
private def withRunner[T](op: Runner => T): T = {
val runner = getRunner()
val result = op(runner)
freeRunner(runner)
result
}
private def killAll(): Unit = allRunners.foreach(_.kill())
// On shutdown, we need to kill all runners:
sys.addShutdownHook(killAll())
// If for some reason the test runner (i.e. sbt) doesn't kill the VM, we
// need to clean up ourselves.
SummaryReport.addCleanup(killAll)
}
}
|
