HyukjinKwon commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4233007629
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,91 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket
directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got
$workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input),
emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should
terminate the launcher")
+ workers.foreach { worker =>
+ assertWorkerStopped(worker, "session close should stop the launched
worker")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got
$workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ assertWorkerStopped(worker, "SIGKILL of the launcher should stop worker")
+ } finally {
+ session.close(emptyCancel)
+ }
+ }
+
+ // An orphaned worker is reparented, and an unreaped zombie stays isAlive
+ // until PID 1 reaps it. State Z in /proc means the worker has already
exited.
+ private def assertWorkerStopped(worker: ProcessHandle, what: String): Unit =
{
Review Comment:
**Nit (P3):** `assertWorkerStopped`, `hasExited`, and `linuxState` land
between the SIGKILL test and `the channel overrides the authority derived from
the socket path`. Every other private helper in this suite (`echoRunner`
through `grpcChannel`, lines 73-122) sits above the first test. The
repository's AGENTS.md asks new members to follow the file's existing
sectioning rather than be placed at their first use. Moving these three
helpers, with their comment, up next to `grpcChannel` keeps the test section
contiguous.
**Verification:**
- **Inspection:** The suite's member order is fields and private helpers
first, then test cases, with no helper definition between tests.
- **Behavior:** Both launcher tests and the existing dispatcher integration
tests still pass after the relocation.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala:
##########
@@ -125,7 +128,11 @@ object EchoGrpcWorkerMain {
server.start()
stderrln(s"listening on $socketPath (transport=${transport.name})")
- // Park until the shutdown hook fires.
+ // SIGKILL skips JVM shutdown hooks. Register before parking so a parent
+ // that has already exited still runs cleanup and unblocks the wait
below.
+ if (launcher.isPresent) {
+ launcher.get().onExit().thenRun(() => cleanup())
Review Comment:
**Nit (P3):** With this registration the worker also shuts down when its
parent process exits, but the object scaladoc at lines 40-41 still says it
"blocks until the JVM receives `SIGTERM` (then shuts down gracefully)".
`ForwardingWorkerMain`'s scaladoc now sends readers here for the parent watch
that covers a launcher SIGKILL. A direct spawn also exits now when the spawning
JVM exits. The class doc should name both stop triggers, for example: "blocks
until the JVM receives `SIGTERM` or its parent process exits (then shuts down
gracefully)".
**Verification:**
- **Inspection:** Reading the object scaladoc against main() shows that
every documented stop path (the SIGTERM shutdown hook and the parent onExit
cleanup) exists in the code, and every stop path in the code is documented.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,91 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket
directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got
$workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input),
emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should
terminate the launcher")
+ workers.foreach { worker =>
+ assertWorkerStopped(worker, "session close should stop the launched
worker")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got
$workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ assertWorkerStopped(worker, "SIGKILL of the launcher should stop worker")
+ } finally {
+ session.close(emptyCancel)
+ }
+ }
+
+ // An orphaned worker is reparented, and an unreaped zombie stays isAlive
+ // until PID 1 reaps it. State Z in /proc means the worker has already
exited.
+ private def assertWorkerStopped(worker: ProcessHandle, what: String): Unit =
{
+ val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10)
+ while (System.nanoTime() < deadline && !hasExited(worker)) {
+ Thread.sleep(50)
+ }
+ if (!hasExited(worker)) {
+ worker.destroyForcibly()
+ fail(s"$what ${worker.pid}")
+ }
+ }
+
+ private def hasExited(worker: ProcessHandle): Boolean = {
+ !worker.isAlive || linuxState(worker.pid).contains('Z')
+ }
+
+ private def linuxState(pid: Long): Option[Char] = {
+ val stat = Paths.get(s"/proc/$pid/stat")
+ if (!Files.isRegularFile(stat)) {
+ None
+ } else {
+ val text = Files.readString(stat)
Review Comment:
**Nit (P3):** If PID 1 reaps the inner worker between
`Files.isRegularFile(stat)` and this read, `/proc/<pid>` disappears and
`readString` throws `NoSuchFileException` (or an ESRCH `IOException` if the
file was opened just before). Nothing catches it, so `assertWorkerStopped`
errors out exactly when the worker has stopped. I checked on OpenJDK 11.0.32
that once a pid is reaped, `isRegularFile` on its stat file returns false and
`readString` throws `NoSuchFileException`. The window is small for each 50 ms
poll, but both launcher tests go through this helper, so it can flake on a busy
Linux CI host. Treating a failed read as "no state" lets the next poll's
`isAlive` check end the wait, for example:
```scala
val text = try Files.readString(stat) catch { case _: IOException => "" }
```
The existing `close < 0` check then returns `None`. This needs
`java.io.IOException` imported in the suite.
**Verification:**
- **Regression:** Calling the state helper for a pid whose /proc entry has
disappeared, such as a process that has just been reaped, returns without
throwing, and the exit check then reports the process as exited.
- **Behavior:** Both launcher tests still pass on Linux, and a still-running
inner worker still makes the wait fail after it is force-killed.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/ForwardingWorkerMain.scala:
##########
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.udf.worker.grpc.testing
+
+import java.io.IOException
+import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
+import java.nio.file.{Files, Paths}
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Test-only launcher that stands in for a sandbox or container launcher. It
+ * receives the engine's `--id` and `--connection`, starts the real worker on a
+ * different socket, and forwards each engine connection to it. This is the
+ * shape a launcher has when the worker cannot bind the engine's socket itself:
+ *
+ * java -cp <classpath> ...ForwardingWorkerMain <worker command...>
+ * --id <id> --connection <uds-path>
+ *
+ * The inner worker exits when this launcher exits, including by `SIGKILL`.
Review Comment:
Thanks, I checked c30ddb05. The scaladoc now says the shutdown hook sends
SIGTERM to the inner worker, waits up to 2 seconds, and then uses SIGKILL. It
also says a SIGKILL of the launcher runs no hook, so the inner worker stops
because EchoGrpcWorkerMain watches its parent. That matches the hook and the
parent watch, so this thread is resolved.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4231761207","thread_id":"inline:4231761207","verdict_sha256":"58d0930295ab8ce0a8053cf959aba9caa5eb64d8fd8eb1b0a3567d4389bbb291"}
-->
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,65 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket
directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got
$workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input),
emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should
terminate the launcher")
+ workers.foreach { worker =>
+ worker.onExit().get(10, TimeUnit.SECONDS)
+ assert(!worker.isAlive, s"session close should stop the launched worker
${worker.pid}")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got
$workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ worker.onExit().get(10, TimeUnit.SECONDS)
Review Comment:
Thanks, I checked c30ddb05. assertWorkerStopped now treats state Z as exited
and calls destroyForcibly before failing, which fixes the zombie timeout and
the leaked worker. One small gap remains in linuxState: if PID 1 reaps the
worker between Files.isRegularFile and Files.readString, readString throws
NoSuchFileException (I confirmed this on OpenJDK 11 for a reaped pid), and the
test fails just as the worker stops. Treating an IOException from that read as
exited, or as None followed by another isAlive check, closes it.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4231761219","thread_id":"inline:4231761219","verdict_sha256":"58d0930295ab8ce0a8053cf959aba9caa5eb64d8fd8eb1b0a3567d4389bbb291"}
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
