zhangfengcdt opened a new issue, #5664:
URL: https://github.com/apache/datafusion-comet/issues/5664
### Describe the bug
#4734 (fix for #4725) makes the global tokio runtime releasable and calls
release_runtime() from CometDriverPlugin.shutdown() /
CometExecutorPlugin.shutdown(). That fixes the hang whenever
SparkContext.stop() runs. It does not fix the case where an application simply
returns from main (or a PySpark driver exits) without calling spark.stop() —
which is a very common shape for batch jobs. In that case the JVM still hangs
forever in DestroyJavaVM, exactly as described in #4725.
The reason is an ordering problem that #4734 cannot get around on its own:
1. Tokio runtime threads that call back into the JVM (memory pool
acquireMemory, ScanExec fed from a JVM iterator, CometScalarSubquery, …) attach
lazily and permanently via jni-rs, using AttachCurrentThread. That makes them
non-daemon JVM threads.
2. When main returns, the launcher calls DestroyJavaVM. HotSpot's
Threads::destroy_vm() waits until it is the last non-daemon thread before it
runs shutdown hooks (invoke_shutdown_hooks() comes after the wait).
3. Spark's ShutdownHookManager is what would call SparkContext.stop() →
plugin shutdown() → release_runtime(). So the runtime can only be released by a
hook that can only run after the runtime's threads have exited. Circular; the
JVM waits forever.
Without an explicit SparkContext.stop() the release in #4734 is unreachable.
### Steps to reproduce
The precondition is simply "at least one tokio runtime thread has called
into the JVM", which any workload with JVM callbacks from spawned work
satisfies. Sketch (upstream-only triggers):
```
// spark-submit / plain `java` main; Comet plugin + native exec enabled;
// spark.memory.offHeap.enabled=true (unified pool → JVM acquireMemory
callbacks)
val spark = SparkSession.builder().master("local[2]")
.config("spark.plugins", "org.apache.spark.CometPlugin")
.config("spark.sql.extensions",
"org.apache.comet.CometSparkSessionExtensions")
.config("spark.comet.enabled", "true")
.config("spark.comet.exec.enabled", "true")
.config("spark.memory.offHeap.enabled", "true")
.config("spark.memory.offHeap.size", "256m")
.getOrCreate()
spark.range(1000000).write.mode("overwrite").parquet(path)
val df = spark.read.parquet(path)
.selectExpr("sum(id)")
.where("1 <= (SELECT max(id) FROM parquet.`" + path + "`)") // scalar
subquery → JVM call from native
df.collect()
require(df.queryExecution.executedPlan.toString.contains("Comet"))
println("main returning")
// intentionally NO spark.stop()
```
Run it as a real process and wait: the process prints main returning and
never exits. jstack shows DestroyJavaVM waiting plus N unnamed non-daemon
threads ("Thread-30", "Thread-31", … — one per runtime worker) with no Java
frames; /proc/<pid>/task/*/comm identifies them as tokio-rt-worker.
### Expected behavior
A Spark application that returns from main without SparkContext.stop()
exits, the same way it does with Comet disabled. Spark's own thread pools are
all daemon precisely so that this works; Comet's runtime threads are currently
the exception.
### Additional context
- Comet main after #4734 (1.0.0); jni 0.22.4; observed on Linux (JDK 17,
Spark 4.1) and macOS (JDK 11, Spark 3.5).
- The upstream development.md threading section already documents that
attachment is cached until the worker exits and that "the tokio runtime has to
be shut down for the JVM to be able to exit" — this issue is the case where
that shutdown can never be initiated.
--
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]
