comphead commented on PR #6121:
URL:
https://github.com/apache/datafusion-comet/pull/6121#issuecomment-5783729435
Checked the redraw against the sources. Separating region from authority is
the right axis and worth landing. Four things I would change first.
**1. The reservation-ceiling label is false for one pool type and imprecise
for the other.**
`parse_memory_pool_config` gives `greedy_unified` a `pool_size` of `0`
(`native/core/src/execution/memory_pools/config.rs:57-62`) and
`CometUnifiedMemoryPool` never reads a limit, so
`spark.comet.exec.memoryPool.fraction` has no effect there at all. Under the
default `fair_unified` the check is `pool_size / num_consumers` against the
pool's *total* used, not the requester's own (`fair_pool.rs:149-163`), so the
ceiling is a fraction of a fraction.
Neither is usually the binding constraint.
`ExecutionMemoryPool.acquireMemory` caps a task at `maxPoolSize /
numActiveTasks` and parks it below `poolSize / (2 * numActiveTasks)`. With 8
concurrent tasks a Comet plan gets roughly 1/8 of the off-heap pool, not all of
it.
**2. `COBJ --> ONPOOL` is a new inaccuracy.** The unified memory manager
accounts only for what consumers explicitly reserve plus cached block sizes.
Comet's plans, `CometVector`s and iterators are ordinary heap objects in user
memory, the `1 - spark.memory.fraction` slice. The old undifferentiated `HEAP`
box was vaguer but less wrong.
**3. The single `OFFPOOL` box hides the execution/storage split.**
`spark.memory.offHeap.size` is divided by `spark.memory.storageFraction` into
off-heap execution and off-heap storage pools
(`UnifiedMemoryManager.scala:64,98-102`). Comet only touches the execution
half, and reaches the rest only by evicting `OFF_HEAP` cached blocks.
**4. Scope is missing.** The pool, the reservation and the fair share are
all per task attempt (`task_shared.rs`), but the diagram is drawn per
container, so a reader cannot see that `spark.executor.cores / spark.task.cpus`
of these run at once.
Two things that predate this PR but sit next to the diagram:
- `memory_management.md` still says to compare `jemalloc_allocated` against
"the summed `thread_NNN_comet_memory_reserved` values". Since #5934 there is an
`alloc-accounting` feature emitting `native_allocated`
(`native/core/src/alloc_accounting.rs`), and `jni_api.rs:252-255` warns that
the per-thread counters must *not* be summed, because a shared pool reports its
full reservation on every thread that references it.
`comet_memory_reserved_total` is the figure to use. `tracing.md` has this
right, and the "No signal for real native usage" open problem is stale for the
same reason.
- Each `createPlan` builds its own `RuntimeEnv` (`jni_api.rs:618,793`) and
therefore its own `FileMetadataCache`, capped at DataFusion's 50 MiB default. A
shuffle runs two plans per task, so eight concurrent tasks can hold up to
roughly 800 MiB of Parquet metadata that no pool sees. Worth a node.
<details>
<summary>An alternative that covers the above, if useful</summary>
```mermaid
%%{init: {'flowchart': {'wrappingWidth': 400}}}%%
flowchart LR
subgraph CG["Executor container, cgroup memory.max = executor.memory +
executor.memoryOverhead + memory.offHeap.size"]
direction TB
subgraph NAT["Native heap, allocated by Rust in this process"]
direction TB
NRES["Declared reservations<br>ExternalSorter, grouped aggregate, hash
and SMJ joins, ShuffleRepartitioner<br>seen by the pool and by
native_allocated"]
NUND["Undeclared Rust allocations<br>expression kernels, array
builders, Parquet page and decompression buffers,<br>per-plan FileMetadataCache
at 50 MiB each, object_store, tokio, FFI batches in flight<br>seen only by
native_allocated"]
NOPAQ["Outside Rust's GlobalAlloc<br>libzstd, aws-lc-sys, libhdfs,
jemalloc retained pages, fragmentation, Arrow padding<br>seen by nothing but
RSS"]
end
subgraph OFF["JVM off-heap, allocated by Unsafe and Java Arrow"]
direction TB
TUNG["Spark Tungsten pages"]
JSH["Comet shuffle pages<br>CometUnifiedShuffleMemoryAllocator"]
ARW["CometArrowAllocator<br>RootAllocator with no limit"]
end
subgraph HEAP["JVM heap, bounded by spark.executor.memory"]
direction TB
UNIF["Spark unified region<br>spark.memory.fraction, default 0.6"]
USER["User memory, the remaining 0.4<br>Comet plans, CometVectors,
iterators, Spark internals"]
end
subgraph REST["Neither heap"]
direction TB
NONHEAP["JVM non-heap<br>metaspace, code cache, JVM and tokio thread
stacks, Netty"]
PAGEC["Page cache from spill and shuffle files<br>charged to the
cgroup, reclaimable under pressure"]
end
end
GATE1["Comet gate, fair_unified only<br>rejects when pool total plus
request exceeds memory_limit / num_consumers<br>memory_limit is
memory.offHeap.size scaled by
spark.comet.exec.memoryPool.fraction<br>greedy_unified has no such gate"]
GATE2["Spark gate, one per task
attempt<br>TaskMemoryManager.acquireExecutionMemory<br>grants at most pool /
numActiveTasks, blocks below pool / 2 x numActiveTasks"]
OFFEX["Off-heap EXECUTION pool<br>one per executor, shared by every task
attempt"]
OFFST["Off-heap STORAGE pool<br>OFF_HEAP cached blocks,
spark.memory.storageFraction"]
ONPOOL["On-heap execution and storage pools"]
GCONLY["Bounded by the heap, budgeted by nobody"]
NOONE["No bound and no budget<br>spark.executor.memoryOverhead is the only
slack"]
NRES -->|"try_grow over JNI. One way only: NativeMemoryConsumer.spill
returns 0"| GATE1
GATE1 --> GATE2
TUNG --> GATE2
JSH --> GATE2
GATE2 --> OFFEX
OFFEX <-->|"borrows, evicting cached blocks"| OFFST
UNIF --> ONPOOL
USER --> GCONLY
NUND --> NOONE
NOPAQ --> NOONE
ARW --> NOONE
NONHEAP --> NOONE
PAGEC --> NOONE
classDef budgeted fill:#c7ecd0,stroke:#2f6b46,color:#000
classDef gate fill:#fff2b2,stroke:#8a7420,color:#000
classDef heaponly fill:#cfe4fb,stroke:#2f5680,color:#000
classDef unbudgeted fill:#f6c69a,stroke:#8a4b1f,color:#000
class NRES,TUNG,JSH,OFFEX,OFFST,UNIF,ONPOOL budgeted
class GATE1,GATE2 gate
class USER,GCONLY heaponly
class NUND,NOPAQ,ARW,NONHEAP,PAGEC,NOONE unbudgeted
```
</details>
Two gate nodes put the fraction and the `1/numActiveTasks` divisor where
they apply, the off-heap pool becomes two boxes with a borrow edge, the heap
separates the unified region from user memory, and the native heap is graded by
which counter can observe it, so the picture doubles as a measurement guide.
The one-way relationship rides on an edge label rather than needing an absent
arrow.
If that is too much for one figure, the natural split is to move the gates
and the pool pair into a separate "what a single reservation passes through"
diagram under *Where Comet's budget comes from*, and leave this one
region-to-authority only.
I have not rendered the fence. `dev/ci/check-mermaid.py` is the authority
and it does not run on my machine either.
--
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]
