dwsmith1983 commented on code in PR #5613:
URL: https://github.com/apache/datafusion-comet/pull/5613#discussion_r4362186288
##########
native/core/src/execution/memory_pools/spark_memory.rs:
##########
@@ -58,34 +56,81 @@ impl SparkMemoryManager for JniMemoryManager {
/// such as a spilled batch read back from disk. When Spark grants less than
[`Self::acquire`]
/// asked for, the shortfall is carried as overcommit rather than failing.
[`Self::release`] repays
/// it before returning anything to Spark, so Spark is never handed back more
than it granted.
-/// While any is outstanding, [`Self::try_acquire`] also asks Spark for it, so
a pool refuses
-/// `try_grow` until Spark can cover both the request and the debt, and the
operator spills.
+/// While any is outstanding, [`Self::try_acquire`] also asks Spark for the
part no other call is
+/// asking for, so a `try_grow` that carries debt is refused unless Spark
covers both, and the
+/// operator spills.
+///
+/// A release while a claim is parked in Spark leaves that request too large,
so it may come back
+/// short and be refused. Handing the claim to Spark on release instead would
break the invariant
+/// below whenever the claim is refused.
///
/// # Invariant
///
/// Every byte a pool records through this type is backed either by Spark's
grant or by
/// overcommit, so Spark's grant plus `overcommit` equals the bytes recorded
and not yet released.
/// Spark is handed back more than it granted only if a release takes less
from `overcommit` than
-/// it could. `CometUnifiedMemoryPool` calls in here from several threads
without a lock, and
-/// updates its own `used` separately, so this rests on three things:
+/// it could. `CometUnifiedMemoryPool` and `CometFairMemoryPool` both call in
here from several
+/// threads without holding a lock, and each updates its own `used`
separately, so this rests on
+/// three things:
///
/// - `overcommit` only grows by bytes that are being recorded in the same
call.
-/// - Each repayment takes its share of `overcommit` in a single atomic
update, so two concurrent
+/// - Each repayment takes its share of `overcommit` under the ledger's lock,
so two concurrent
/// calls can never repay the same debt.
/// - A caller never releases more than it recorded. DataFusion guarantees
this, because a shrink
/// cannot exceed the reservation it comes from.
pub(super) struct SparkMemory {
manager: Box<dyn SparkMemoryManager>,
- overcommit: AtomicUsize,
+ /// Held only for arithmetic, never across a call to Spark.
+ ledger: Mutex<Ledger>,
task_attempt_id: i64,
}
+#[derive(Default)]
+struct Ledger {
+ /// Bytes recorded without Spark's grant.
+ overcommit: usize,
+ /// Debt that `try_acquire` calls in flight have put into their Spark
requests. A release
+ /// can repay that debt meanwhile, so this can exceed `overcommit`.
+ claimed: usize,
+}
+
Review Comment:
> Could the ledger and `Claim` move to their own PR, with an issue for the
unified pool?
Yes. `spark_memory.rs` and `unified_pool.rs` are back to main here, the
unified pool's race is #7174, and the ledger will get its own PR.
--
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]
