andygrove opened a new pull request, #5993:
URL: https://github.com/apache/datafusion-comet/pull/5993
## Which issue does this PR close?
Relates to #4576. Draft, opened for discussion rather than merge, and
deliberately overlapping with @comphead's #5666 so the two approaches can be
compared.
## Rationale for this change
When an executor exceeds its container memory limit the kernel kills the
whole JVM, taking every task on it, its cached blocks and its shuffle files. A
single failed task is retried by Spark and costs almost nothing by comparison.
This adds an optional guard that samples the container's real memory usage and
fails the current task when usage approaches the limit.
The design differs from the two previous attempts in one respect, and it is
the point of the PR: **it reads the kernel's number rather than Comet's own
accounting.**
The closing review of #4582 set this precondition:
> Whether to enforce on that number at all is a decision I'd rather make
after seeing how well it tracks RSS on real workloads, rather than before.
I measured that and posted the result on #4576. On TPC-H SF100 with the
default allocator, across ~4500 samples pairing `native_allocated` with kernel
RSS at the same trace anchor:
- delta correlation 0.42 to 0.44
- the balance never leads RSS: correlation within 0.01 of zero at 1, 2, 5
and 10 sample horizons
- the gap between them wanders by 2 to 3 GB, while the balance itself only
spans 0 to 2.1 GB
A threshold on the allocator balance cannot mean anything about the quantity
the kernel kills on, because the noise in the offset exceeds the whole signal.
That is the same conclusion the #4582 review reached analytically ("the tracked
balance is layout bytes, not RSS ... that gap is unbounded and always in the
dangerous direction"), now with numbers.
Reading the cgroup avoids the problem entirely: it is the number the OOM
killer compares against the limit, and it already includes the JVM heap,
Comet's native allocations, JVM-side Arrow, Spark's own off-heap and mapped
files. In my measurements the JVM heap was the dominant term, so any signal
that excludes it is measuring the wrong thing.
## What changes are included in this PR?
- `native/core/src/execution/memory_guard.rs` (new). Discovers the
container's memory usage file and limit: cgroup v2 at the mount root (the
Kubernetes case, where the pod's cgroup is namespaced), then cgroup v2 resolved
from `/proc/self/cgroup` for non-containerised hosts, then cgroup v1 for older
Kubernetes and YARN. `check()` samples usage, throttled to at most once per 100
ms, and reports a trip at a configurable fraction of the limit.
- `jni_api.rs` calls it at the two existing execution checkpoints: the async
`batch_receiver` path and the ScanExec busy-poll path's 100-poll checkpoint. A
trip becomes `DataFusionError::ResourcesExhausted`, which reaches the JVM as
`CometNativeException`, which `CometExecIterator` logs with the task id and
rethrows, so Spark fails and retries one task.
- `spark.comet.exec.memoryGuard.enabled` (default `false`) and
`spark.comet.exec.memoryGuard.threshold` (default `0.9`). Names match #5666 so
the two are comparable.
- Tuning guide section.
No allocator wrapper, no `panic_any`, no stamped-thread set. That sidesteps
the four defects the #4582 review found: the two enforcement layers not
actually layering, the stamped set covering tokio's blocking pool so panics
escaped as `JoinError`, the `LOCAL_DRIFT` leak on thread exit, and the
layout-bytes-versus-RSS gap.
## Limitations, stated up front
- **Checks happen between batches.** An operator that grows sharply within a
single batch can still cross the limit before the guard observes it. This
narrows the window, it does not close it. Closing it entirely means acting
inside the allocator, which #4582 tried and which brings reentrancy and
unwinding problems of its own.
- **It fails the task rather than spilling first.** Spilling releases
reserved bytes, and in measurements on #5983 the overshoot lived in allocations
that never reserved, so spill-then-retry ran into the same wall and only
delayed the failure.
- **No per-task attribution.** The cgroup counter is process-wide, so the
task that gets failed is the one that happened to reach a checkpoint, not
necessarily the one responsible.
- **Not validated in a container.** See testing below. This is the main
reason it is a draft.
## How are these changes tested?
Unit tests for the parsing, which is the part that can be tested off Linux:
the cgroup v2 path from `/proc/self/cgroup` including the `0::/` container
form, a v1-only host producing no v2 path, `max` treated as no limit, usage
parsing, and out-of-range thresholds disabling the guard.
**What is not tested, and I would not merge it without this:** the guard has
never run in a container. `discover()` is gated on a runtime `cfg!(target_os =
"linux")` check, so the `/sys/fs/cgroup` reads typecheck on macOS but never
execute there, and I have no Kubernetes environment to hand. What it needs is a
pod with a memory limit, a query that exceeds it, and confirmation that the
task fails and is retried while the executor survives. If anyone can run that,
I would value it more than any amount of further code review.
--
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]
