andygrove commented on issue #1943:
URL:
https://github.com/apache/datafusion-ballista/issues/1943#issuecomment-4886371492
## Reference: how Spark's shuffle fetch avoids all three corners
For whoever picks up the fix — Spark solves this exact problem, and its
design points
at what Ballista is actually missing. Verified against the Spark 3.5.8 source
(`org.apache.spark.storage.ShuffleBlockFetcherIterator`,
`org.apache.spark.network.util.TransportConf`).
**Transport.** Not gRPC/HTTP-2 — a custom length-delimited Netty TCP protocol
(`NettyBlockTransferService` / `TransportClientFactory`). There is **no
protocol-level
64 KB connection-level flow-control window** to starve, which is what
deadlocked the
multiplexing attempts here.
**Connections.** `spark.shuffle.io.numConnectionsPerPeer` defaults to **1**
(`TransportConf.numConnectionsPerPeer()` → `1`). All block fetches to a peer
are
**multiplexed over that one long-lived channel** — i.e. Spark runs the exact
model that
hung Ballista at 1 connection/peer, and it is fine.
**Why it doesn't deadlock — a reduce-side in-flight governor.** This is the
piece
Ballista lacks. `ShuffleBlockFetcherIterator` only issues a fetch when
(`isRemoteBlockFetchable`, SBFI.scala:1215):
```scala
bytesInFlight == 0 ||
(reqsInFlight + 1 <= maxReqsInFlight &&
bytesInFlight + fetchReqQueue.front.size <= maxBytesInFlight)
```
with, by default:
- `spark.reducer.maxSizeInFlight` = **48m** (`maxBytesInFlight`),
- `targetRemoteRequestSize = max(maxBytesInFlight / 5, 1)` — requests are
pre-split so up
to ~5 peers fetch in parallel (SBFI.scala:112),
- `spark.reducer.maxReqsInFlight` (default `Int.MaxValue`) and
`spark.reducer.maxBlocksInFlightPerAddress` — caps on outstanding requests
/ blocks
per address.
So the amount of concurrent in-flight data on any connection is bounded **at
the
application layer**. The transport is never flooded, so one multiplexed
connection is
enough and never starves. Ballista opens a `do_get` stream per partition
with **no global
in-flight-bytes/requests governor**, so it must either flood one h2
connection (→ window
deadlock) or open a connection per fetch (→ churn).
**Resilience.** Shuffle fetches are discrete, idempotent **blocks**, and
Spark retries
them: `spark.shuffle.io.maxRetries` = **3**
(`TransportConf.maxIORetries()`). A mid-fetch
connection break just refetches the block. Large blocks stream **to disk**
(`spark.maxRemoteBlockSizeFetchToMem`) instead of pinning memory/the
connection. Contrast
Ballista, where `execute_do_get`'s retry covers only the request + schema
message, so a
broken pipe during the body is fatal, and a partition is an open-ended
stream rather than
a bounded, retriable unit.
### Mapping to the three-corner tension above
| Ballista failure | How Spark avoids it |
|---|---|
| Multiplex → h2 64 KB window deadlock | Netty transport, no shared protocol
window; **plus** reduce-side in-flight cap so a connection is never overloaded |
| Exclusive-per-stream → churn / pool starvation | 1 reused connection/peer,
many blocks multiplexed, throttled by in-flight bytes/requests |
| Broken pipe mid-stream fatal | Blocks are idempotent + retried
(maxRetries=3); big blocks spill to disk |
### Design takeaway
The decisive thing Spark has that Ballista doesn't is a **reduce-side
in-flight governor**
(`maxBytesInFlight` / `maxReqsInFlight` / `maxBlocksInFlightPerAddress`).
That governor is
what makes *multiplexing over very few connections* safe. So the faithful
fix is not a
pool-shape tweak or just larger h2 windows in isolation, but roughly:
1. a small, reused set of connections per peer (even 1–N), **plus**
2. a bounded in-flight-bytes / in-flight-requests governor on the shuffle
reader, **plus**
3. retriable, bounded block fetches (retry the body phase; optionally spill
large
partitions to disk).
That combination dissolves all three corners at once, and it's a direct port
of a model
that's been load-bearing in Spark for years.
--
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]
