andygrove commented on issue #1943:
URL:
https://github.com/apache/datafusion-ballista/issues/1943#issuecomment-4886318132
## Investigation summary: root cause + six attempts + the design tension
I spent a good while reproducing and instrumenting this on a 2-executor ×
8-core
cluster (TPC-H SF100 from S3/MinIO, `target_partitions` 128/256). Posting
the full
findings so the fix can be decided as a design question rather than guessed
at — I
tried six approaches and each hit a different corner of what looks like a
fundamental
tension.
### Confirmed root cause: shuffle-connection churn
With executor-side `RUST_LOG=info,h2=debug`, a single failing
`target_partitions=128`
run shows the shuffle client **opening and tearing down ~3,500 connections**
(client-
initiated `GOAWAY` frames, all `NO_ERROR`; **zero `RST_STREAM`**, so not h2
rapid-reset). The failure is an intermittent **race**: the client's own
connection
close racing an in-flight body read produces the `h2 … BrokenPipe` ("error
reading a
body from connection"); with client caching off the same churn exhausts
ephemeral
ports (`EADDRNOTAVAIL`). The churn comes from the pool checking out an
**exclusive
connection per fetch** and returning it on drop.
### The tension (why it's not a one-line fix)
A shuffle read opens **many concurrent partition streams**. Each open stream
needs
*either* its own connection *or* to share one — and every option hits a wall:
| Connection model | Failure |
|---|---|
| One connection **per fetch** (today) | churn → broken-pipe race /
ephemeral-port exhaustion |
| Hold-connection-for-stream + keepalive | still broken pipe — the churn is
unchanged |
| **Multiplex**, 1 connection/peer | **hang** — the shared 64 KB h2
connection-level flow-control window starves under many concurrent fetch
streams; executors go fully idle |
| **Multiplex**, small pool (N/peer) | same hang, just on heavier queries
(raises the threshold, doesn't remove it) |
| **Bounded exclusive pool** (semaphore) | **hang** — permit starvation: a
stage opens more concurrent partition streams than the pool cap; the streams
holding connections can't release (they're waiting on the consumer, which is
waiting on a stream that can't get a connection) |
| Unbounded exclusive pool | back to churn / port exhaustion |
So: exclusive-per-open-stream → either unbounded (churn/ports) or bounded
(starvation
deadlock); shared/multiplexed → h2 flow-control deadlock. Each of the six
attempts
landed on one of these.
### What each attempt showed (evidence)
1. **Hold the pooled connection for the response-stream lifetime** (so it
isn't
reused/evicted mid-stream). Reduced it but did **not** fix it — still
broke, because
the *churn* itself remained.
2. **HTTP/2 keepalive on shuffle connections** (they were created with a
`None` config,
i.e. *no* keepalive at all). Correct on its own merits, but didn't fix
the broken
pipe — the connection dies while *actively* streaming, not while idle.
3. **Multiplex over one connection per peer.** Removed the churn entirely,
but **hung**:
the executors sit at ~0 CPU on the first shuffle-heavy query. The tiny
default
connection-level flow-control window is the suspect.
4. **Multiplex over a small round-robin pool (N=8 per peer).** Cleared the
moderate
shuffle stage but **hung on a heavier join** (more concurrent
streams/connection).
5. **Bounded *exclusive* pool (semaphore, ≤64/endpoint, reuse, one fetch per
connection).** Also **hung**, and earlier — permit starvation as above.
(Per-query timings, where runs completed, were on par with `main` — so none
of this is
a throughput problem; it's a liveness one.)
### Options that remain (haven't tried / need design)
1. **Multiplex + enlarge the h2 flow-control windows.** Bump
`initial_connection_window_size` / `initial_stream_window_size` from the
64 KB
default to several MB on the shuffle client (and matching server
settings). This is
the one principled lever I did **not** pull, and it directly targets the
multiplex
deadlock cause. If it works, multiplexing becomes viable and removes the
churn.
2. **Fetch each partition to completion before releasing the connection**
(buffer to
memory or spill to disk). Decouples connection-hold-time from
consume-time, so a
bounded exclusive pool no longer starves. Costs a per-partition buffer.
3. **A dedicated shuffle transport** not built on per-partition Flight
`do_get`.
### Environment / repro
- Ballista `main` (DataFusion 54), executors 8 cores each, `--client-ttl 60`.
- TPC-H SF100 from S3; `target_partitions` 128 fails ~2 of 3 runs on `main`;
256 fails
more often. The shuffle-heavy queries (Q7/Q9/Q17/Q21) are where failures
cluster.
Happy to run any of the above against this cluster if a direction is chosen.
My instinct
is option 1 (flow-control windows) is the smallest principled next step, but
this is a
core-transport decision so I wanted the full picture on the record first.
--
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]
