[SPARK-58021][CONNECT] Add an opt-in pool of pre-warmed, single-use local Spark Connect servers#57102
Draft
ericm-db wants to merge 11 commits into
Draft
[SPARK-58021][CONNECT] Add an opt-in pool of pre-warmed, single-use local Spark Connect servers#57102ericm-db wants to merge 11 commits into
ericm-db wants to merge 11 commits into
Conversation
…for faster local startup
Adds an opt-in (SPARK_LOCAL_CONNECT_REUSE / spark.local.connect.reuse) so that
`SparkSession.builder.remote("local[*]").getOrCreate()` reconnects to a persistent local
Spark Connect server -- starting a detached one on the first run and reconnecting to it on
later runs -- instead of booting a fresh in-process server in every process. The first run
pays the cold start once; later runs reconnect in a fraction of a second.
Default behavior is unchanged when the opt-in is off. No protocol or Scala changes.
Trim verbose comments/docstrings, drop redundant noqa and the private-method example from the user docs, and consolidate exception handling. No functional change.
…d seed server confs Fixes two issues raised in review of the persistent local Connect server path: 1. Concurrent first-time startup could fail spuriously. Opted-in processes starting at the same time (or after a version upgrade) would each spawn a server on the fixed port; the loser's JVM failed to bind and raised LOCAL_CONNECT_SERVER_START_FAILED even though a usable server was now up. Serialize start-up with a discovery-file lock, reconnect to the winner if our own daemon exits, and fall back to an ephemeral port when the configured one is already held. 2. The daemon dropped most builder options on first startup. Forward the caller's start-up confs (warehouse dir, jars/packages, catalog, app name, etc.) to the server via a JSON conf file so first-run behavior matches the in-process path. A later run reconnecting to an already-warm JVM still cannot change its static confs, as documented.
On the start-up timeout path, terminating only the detached daemon process left its child JVM orphaned when the timeout fired before the daemon had installed its own signal handling (the slow- startup case that triggers the timeout). Signal the whole process group -- the daemon is a session leader and the JVM stays in its group -- and escalate to SIGKILL if a graceful stop does not take, so the JVM is reaped instead of leaking.
Adds a POSIX-only test that starts the detached daemon, waits until it has spawned its child JVM, then calls _terminate_local_connect_server and asserts the whole process group is gone -- covering the orphaned-JVM case the timeout-path fix addresses. Also waits for the server port to close in tearDown so a stopped server's JVM cannot linger into the next test.
…ing the local server The stop path signals the pid recorded in the discovery file via killpg. If that pid is stale (e.g. the daemon was SIGKILLed and left its discovery file behind) and has been recycled by an unrelated process, group-killing it could take down the caller's own process group. The daemon is always launched as a session leader, so its group id equals its pid; only signal the group when that holds, and fall back to a plain kill otherwise. Adds a unit test covering both directions (with the signal syscalls mocked so a regression cannot kill the test runner), and a ruff-format fix for a missing blank line after the local_server.py module docstring. Co-authored-by: Isaac
…ocal Spark Connect servers Adds spark.local.connect.pool / SPARK_LOCAL_CONNECT_POOL: each run claims a fresh, never-used local Connect server from a pre-warmed pool (atomic rename of its discovery file), replacements spawn asynchronously, and the claimed server is torn down asynchronously when the session stops. No JVM ever serves two runs. Builds on the SPARK-57787 daemon and discovery plumbing. Co-authored-by: Isaac
…e pool members Race fix: a cold-pool waiter could conclude every in-flight launch had died and fail, because 'no live pending launches' is a transient observation -- a claim consumes its launch marker an instant before the winner's refill publishes a replacement, and a fresh marker briefly carries a -1 placeholder pid. Markers now record their spawner pid (a fresh placeholder is live while its spawner is), the dry check is debounced across polls, and a persistently dry pool is respawned under the lock; only the deadline fails an acquire. Reproduced with 4 concurrent clients, fixed, and revalidated (8 rounds clean). Warmup: pool members now run fixed synthetic queries (classic pass plus a self-connect pass) after publishing their discovery file, so JVM-global JIT and codegen caches are hot before the member's single real run. First-query latency drops from ~1.4s to ~0.04s; a 16-query run on a warmed fresh server (1.14s) matches a reused server seeing those shapes first (1.21s). Disable with SPARK_LOCAL_CONNECT_POOL_WARMUP=0. Daemon hardening found while testing the above: the self-connect warmup runs in a bounded daemon thread and checks the stop flag between steps (a release mid-warmup SIGTERMs the group; a Connect handshake against the dying JVM otherwise wedges the main thread forever, leaving an unkillable daemon and a zombie JVM); the reaper loop exits when the child JVM process is gone instead of failing open; shutdown survives a dead gateway; and an env-gated phase log (SPARK_LOCAL_CONNECT_SERVER_DEBUG_LOG) makes detached daemons debuggable. Co-authored-by: Isaac
… cannot outlive the process group test_terminate_reaps_daemon_and_jvm fails in CI because the job container's PID 1 does not reap orphans: on group SIGTERM the daemon exits before its child JVM finishes shutting down, the JVM is reparented to PID 1, and its exit leaves a permanent zombie. A zombie still counts as a process-group member, so killpg(pgid, 0) never raises ProcessLookupError -- and no amount of SIGKILL (the previous hardening attempt) can remove it. The daemon now stops Spark on a bounded thread, closes the JVM's stdin pipe (the gateway exits cleanly on stdin EOF), and waits for / kills and reaps the JVM before exiting, so the group always drains to empty. Also replace the star-import __main__ block in test_connect_local_server_pool.py with the pyspark.testing.main convention used by the sibling suites, fixing the ruff F403/RUF100 lint failure. Co-authored-by: Isaac
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Adds a second opt-in fast path for local Spark Connect development, complementing the
persistent-server reuse mode of #56907 (SPARK-57787): a pool of pre-warmed, single-use local
Connect servers.
With
SPARK_LOCAL_CONNECT_POOL=1(or.config("spark.local.connect.pool", "true")),SparkSession.builder.remote("local[*]").getOrCreate():~/.spark/connect-local-pool,override via
SPARK_LOCAL_CONNECT_POOL_DIR) -- a claim is an atomic rename of the server'sdiscovery file, so exactly one racing client wins;
spark.local.connect.pool.size(default 2) memberswarm;
no server ever serves two runs.
Mechanics:
client's start-up confs, and is stamped with a conf fingerprint; a run only claims members whose
fingerprint matches its own confs, so a warm JVM is never handed to a run whose static confs
differ from what it was booted with.
fcntllock serializes spawn accounting so concurrent cold starters share onecomplement of launches instead of each spawning their own; claims themselves are lock-free
atomic renames. Platforms without
fcntlmay transiently over-spawn; extras are reclaimed bythe idle timeout (same degradation posture as the reuse path's start lock).
version-mismatched members, servers claimed by clients that died uncleanly (e.g. SIGKILL), and
retired servers that hang in shutdown (SIGTERM at release, escalated to a hard kill after 30s).
fixed set of synthetic queries (a classic pass, then a self-connect pass exercising the
Connect planner / Arrow / gRPC path), so the one real run each member serves starts with hot
JVM-global JIT and codegen caches. No user code or state is involved, and every member gets
the identical warmup. Disable with
SPARK_LOCAL_CONNECT_POOL_WARMUP=0.--exit-after-use(self-terminate once their oneclient's session has come and gone) and an idle timeout
(
SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT, default 1800s), so an idle machine drains to zero.SPARK_LOCAL_CONNECT_SERVER_DEBUG_LOG=<file>, which logstimestamped lifecycle phases (boot, warmup, signals, shutdown).
SparkSession._purge_local_connect_pool()force-kills every member (warm, in-flight, orclaimed) and empties the pool directory -- a one-command "really clean state" escape hatch.
Relationship to the reuse mode (#56907): reuse keeps one long-lived server that all runs
reconnect to -- cheapest, but state backed by the shared
SparkContext(persistent catalog,global temp views, cached data) carries across runs and the JVM ages. The pool trades standing
memory (~550 MB RSS per warm member measured on macOS with default confs; the default pool of 2
is about 1.1 GB while actively iterating) for per-run isolation identical to today's default
behavior. This directly addresses the determinism concern raised on the dev list about reusing a
long-lived backend server:
https://lists.apache.org/thread/sg9o2gbb3nttz74f0s01v8f167zy8ltt
Stacked on #56907: the earlier commits belong to that PR; only the last commit is new here.
Will rebase once #56907 merges.
Why are the changes needed?
Local
getOrCreate()pays a multi-second JVM +SparkContextcold start in every process.#56907 removes that cost by reusing one long-lived server, but reuse was flagged on the dev list
as a determinism/isolation risk if it ever became a default: a shared backend can be left in a
bad state that no fresh environment reproduces. The pool keeps the latency win while preserving
fresh-server-per-run semantics, making it a candidate for an eventual default-on story.
Measured on a dev laptop (medians;
local[*], default confs). "Time to first result" is whatthe user waits between launching
python script.pyand the first query returning: import +getOrCreate()+ first query. The rows:Connect server in-process and discards it at exit, so every run pays the full cold start plus
the fresh JVM's first-query JIT cost.
SPARK_LOCAL_CONNECT_POOL_WARMUP=0): the run claims a pre-booted,never-used server, so the session comes up fast -- but that JVM has never executed a query,
so the first query still pays the full JIT/codegen cost. Not a recommended configuration;
shown only to isolate what the self-warmup contributes.
itself on synthetic queries while idling in the pool, so the first real query is also fast.
This row is the user experience of this PR.
reconnecting to the one long-lived shared server, hot from previous runs. Included as the
speed ceiling: the fastest any design can be, achieved by sharing a JVM across runs.
The pool and reuse rows assume the once-per-work-session bootstrap already happened (the first
run on a cold pool costs ~2.8-3.0s, a cold reuse start ~2.5s); the default row has no such
footnote because every run is its own bootstrap.
Over a 16-query session (4 shapes x 4 rounds): warmed pool 1.14s total vs 1.21s for a reused
server seeing those shapes for the first time vs 0.75s for a reused server that has executed the
exact same queries before. The fresh-JVM warmup tax is front-loaded and bounded, not per-query.
Does this PR introduce any user-facing change?
Only when the opt-in is enabled. With
SPARK_LOCAL_CONNECT_POOL=1(orspark.local.connect.pool=true), alocalremotegetOrCreate()claims a pre-warmed, never-usedlocal Connect server and replaces it in the background, instead of booting an in-process server.
With the opt-in unset (the default), behavior is unchanged. If both this and
spark.local.connect.reuseare set, the pool takes precedence.How was this patch tested?
New
python/pyspark/sql/tests/connect/test_connect_local_server_pool.py:bookkeeping; janitor rules (dead launches, unreachable members, claims orphaned by SIGKILLed
clients, retired-server escalation, conf-seed cleanup) with live process groups standing in for
servers; release writes the retired marker and stops the server; purge kills all members and
empties the directory; pool lock roundtrip; pool-size parsing.
torn down afterwards; two concurrent cold clients get distinct servers.
"all launches died") was reproduced with instrumented clients, fixed, and revalidated with 8
rounds x 4 concurrent clients, all clean. A teardown hang (release mid-warmup SIGTERMs the
group; the self-connect warmup then wedged against the dying JVM, leaving an unkillable daemon
and zombie JVM) was reproduced via the daemon phase log, fixed (stop-flag checks between warmup
steps, bounded connect-warmup thread, exit-when-JVM-dead in the reaper), and re-verified.
Also verified manually on macOS: two simultaneous cold clients share one spawn complement under
the pool lock; a global temp view and a cached table created in one run are not visible to the
next (the state that leaks by design under the reuse mode); a SIGKILLed client's orphaned server
is reaped by the next run's janitor while that run still connects in ~0.4s; purge leaves zero
pool processes; idle RSS per warm member measured at 534-557 MB.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)