comphead commented on code in PR #6128: URL: https://github.com/apache/datafusion-comet/pull/6128#discussion_r4078608726
########## native/core/src/execution/memory_pools/spark_memory.rs: ########## @@ -0,0 +1,223 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::{ + atomic::{AtomicUsize, Ordering::Relaxed}, + Arc, +}; + +use jni::objects::{Global, JObject}; +use log::warn; + +use crate::{errors::CometResult, jvm_bridge::JVMClasses}; + +/// Spark's side of a Comet pool: the calls that acquire and release off-heap execution memory. +pub(super) trait SparkMemoryManager { + /// Asks Spark for `size` bytes and returns how many it granted. + fn acquire(&self, size: usize) -> CometResult<i64>; + fn release(&self, size: usize) -> CometResult<()>; +} + +/// Calls [`crate::jvm_bridge::CometTaskMemoryManager`] over JNI. +pub(super) struct JniMemoryManager(Arc<Global<JObject<'static>>>); + +impl SparkMemoryManager for JniMemoryManager { + fn acquire(&self, size: usize) -> CometResult<i64> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, + comet_task_memory_manager(handle).acquire_memory(size as i64) -> i64) + }) + } + + fn release(&self, size: usize) -> CometResult<()> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, comet_task_memory_manager(handle).release_memory(size as i64) -> ()) + }) + } +} + +/// Memory a Comet pool holds from Spark, including any it has recorded without Spark's grant. +/// +/// `MemoryPool::grow` must always succeed: DataFusion calls it for memory that already exists, +/// 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. +/// The pool's own `used` still counts the full amount, so its next `try_grow` is refused. Review Comment: "The pool's own `used` still counts the full amount, so its next `try_grow` is refused." That holds for `CometFairMemoryPool`, whose `try_grow` compares `state.used` against `pool_size / num`. I do not think it holds for `CometUnifiedMemoryPool`. That `try_grow` never reads `used`. It asks Spark, and Spark's ledger only contains what it actually granted, so it has no idea the overcommit exists. Under `greedy_unified` a `grow` shortfall would then produce no back-pressure at all, and successive shortfalls compound with nothing to stop them short of the cgroup. Could the unified `try_grow` fold the outstanding debt into the request, asking Spark for `additional + overcommit()` and repaying the debt out of the grant first? If you would rather keep that out of scope, could the doc say plainly that `greedy_unified` leans on Spark alone? The same sentence is echoed on the `grow` doc comments in both pools, so all three want the same correction. ########## native/core/src/execution/memory_pools/spark_memory.rs: ########## @@ -0,0 +1,223 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::{ + atomic::{AtomicUsize, Ordering::Relaxed}, + Arc, +}; + +use jni::objects::{Global, JObject}; +use log::warn; + +use crate::{errors::CometResult, jvm_bridge::JVMClasses}; + +/// Spark's side of a Comet pool: the calls that acquire and release off-heap execution memory. +pub(super) trait SparkMemoryManager { + /// Asks Spark for `size` bytes and returns how many it granted. + fn acquire(&self, size: usize) -> CometResult<i64>; + fn release(&self, size: usize) -> CometResult<()>; +} + +/// Calls [`crate::jvm_bridge::CometTaskMemoryManager`] over JNI. +pub(super) struct JniMemoryManager(Arc<Global<JObject<'static>>>); + +impl SparkMemoryManager for JniMemoryManager { + fn acquire(&self, size: usize) -> CometResult<i64> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, + comet_task_memory_manager(handle).acquire_memory(size as i64) -> i64) + }) + } + + fn release(&self, size: usize) -> CometResult<()> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, comet_task_memory_manager(handle).release_memory(size as i64) -> ()) + }) + } +} + +/// Memory a Comet pool holds from Spark, including any it has recorded without Spark's grant. +/// +/// `MemoryPool::grow` must always succeed: DataFusion calls it for memory that already exists, +/// 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. +/// The pool's own `used` still counts the full amount, so its next `try_grow` is refused. +pub(super) struct SparkMemory<M = JniMemoryManager> { + manager: M, + overcommit: AtomicUsize, + task_attempt_id: i64, +} + +impl SparkMemory { + pub(super) fn new(handle: Arc<Global<JObject<'static>>>, task_attempt_id: i64) -> Self { + Self::with_manager(JniMemoryManager(handle), task_attempt_id) + } +} + +impl<M: SparkMemoryManager> SparkMemory<M> { + fn with_manager(manager: M, task_attempt_id: i64) -> Self { + Self { + manager, + overcommit: AtomicUsize::new(0), + task_attempt_id, + } + } + + /// Acquires `size` bytes, or none: a partial grant is handed back and reported as `Err` with + /// the number of bytes Spark offered. + pub(super) fn try_acquire(&self, size: usize) -> CometResult<Result<(), usize>> { + let granted = granted(size, self.manager.acquire(size)?); + if granted < size { + self.manager.release(granted)?; + return Ok(Err(granted)); + } + Ok(Ok(())) + } + + /// Acquires what Spark will grant toward `size` bytes and carries the rest as overcommit. + /// Never fails; a failed call to Spark counts as a zero grant. + pub(super) fn acquire(&self, size: usize) { + let granted = match self.manager.acquire(size) { + Ok(acquired) => granted(size, acquired), + Err(e) => { + warn!( + "Task {} failed to acquire {size} bytes from Spark: {e:?}", + self.task_attempt_id + ); + 0 + } + }; + if granted < size { + self.overcommit.fetch_add(size - granted, Relaxed); + } + } + + /// Frees `size` bytes, repaying overcommit before releasing the rest to Spark. + pub(super) fn release(&self, size: usize) -> CometResult<()> { + let mut to_release = size; + if self.overcommit.load(Relaxed) > 0 { + let prev = self + .overcommit + .fetch_update(Relaxed, Relaxed, |debt| Some(debt.saturating_sub(size))) + .unwrap(); + to_release = size.saturating_sub(prev); + } Review Comment: The `load` guard can go. `fetch_update` with `saturating_sub` is already a no-op that returns `Ok(0)` when there is no debt, so the fast path buys nothing ahead of a JNI call, and the read-then-CAS shape makes a reader stop and work out whether it is racy. ```rust let debt = self .overcommit .fetch_update(Relaxed, Relaxed, |debt| Some(debt.saturating_sub(size))) .unwrap(); let to_release = size.saturating_sub(debt); ``` ########## native/core/src/execution/memory_pools/spark_memory.rs: ########## @@ -0,0 +1,223 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::{ + atomic::{AtomicUsize, Ordering::Relaxed}, + Arc, +}; + +use jni::objects::{Global, JObject}; +use log::warn; + +use crate::{errors::CometResult, jvm_bridge::JVMClasses}; + +/// Spark's side of a Comet pool: the calls that acquire and release off-heap execution memory. +pub(super) trait SparkMemoryManager { + /// Asks Spark for `size` bytes and returns how many it granted. + fn acquire(&self, size: usize) -> CometResult<i64>; + fn release(&self, size: usize) -> CometResult<()>; +} + +/// Calls [`crate::jvm_bridge::CometTaskMemoryManager`] over JNI. +pub(super) struct JniMemoryManager(Arc<Global<JObject<'static>>>); + +impl SparkMemoryManager for JniMemoryManager { + fn acquire(&self, size: usize) -> CometResult<i64> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, + comet_task_memory_manager(handle).acquire_memory(size as i64) -> i64) + }) + } + + fn release(&self, size: usize) -> CometResult<()> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, comet_task_memory_manager(handle).release_memory(size as i64) -> ()) + }) + } +} + +/// Memory a Comet pool holds from Spark, including any it has recorded without Spark's grant. +/// +/// `MemoryPool::grow` must always succeed: DataFusion calls it for memory that already exists, +/// 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. +/// The pool's own `used` still counts the full amount, so its next `try_grow` is refused. +pub(super) struct SparkMemory<M = JniMemoryManager> { + manager: M, + overcommit: AtomicUsize, + task_attempt_id: i64, +} + +impl SparkMemory { + pub(super) fn new(handle: Arc<Global<JObject<'static>>>, task_attempt_id: i64) -> Self { + Self::with_manager(JniMemoryManager(handle), task_attempt_id) + } +} + +impl<M: SparkMemoryManager> SparkMemory<M> { + fn with_manager(manager: M, task_attempt_id: i64) -> Self { + Self { + manager, + overcommit: AtomicUsize::new(0), + task_attempt_id, + } + } + + /// Acquires `size` bytes, or none: a partial grant is handed back and reported as `Err` with + /// the number of bytes Spark offered. + pub(super) fn try_acquire(&self, size: usize) -> CometResult<Result<(), usize>> { + let granted = granted(size, self.manager.acquire(size)?); + if granted < size { + self.manager.release(granted)?; + return Ok(Err(granted)); + } + Ok(Ok(())) + } + + /// Acquires what Spark will grant toward `size` bytes and carries the rest as overcommit. + /// Never fails; a failed call to Spark counts as a zero grant. + pub(super) fn acquire(&self, size: usize) { + let granted = match self.manager.acquire(size) { + Ok(acquired) => granted(size, acquired), + Err(e) => { + warn!( + "Task {} failed to acquire {size} bytes from Spark: {e:?}", + self.task_attempt_id + ); + 0 + } + }; + if granted < size { + self.overcommit.fetch_add(size - granted, Relaxed); Review Comment: `jni_api.rs:799` sets `target_partitions` to `spark.task.cpus`, so partitions of the same plan run on different Tokio workers against one task-shared pool. In `CometFairMemoryPool` the `state` mutex serializes the JNI call with the `used` update, so the ledger moves as a single step. In `CometUnifiedMemoryPool` nothing serializes `manager.acquire()` with this `fetch_add`, or either of them with the pool's `used`. I walked through a few interleavings and I believe the guarantee survives, because a `release` can only ever be as large as the reservation the shrinking consumer has already counted in `used`. But that is the guarantee #1732 exists to protect, and it now rests on a three-counter argument that is not written down anywhere. Could you state it as an invariant here, so the next person to touch `acquire` knows what they have to preserve? ########## native/core/src/execution/memory_pools/spark_memory.rs: ########## @@ -0,0 +1,223 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::{ + atomic::{AtomicUsize, Ordering::Relaxed}, + Arc, +}; + +use jni::objects::{Global, JObject}; +use log::warn; + +use crate::{errors::CometResult, jvm_bridge::JVMClasses}; + +/// Spark's side of a Comet pool: the calls that acquire and release off-heap execution memory. +pub(super) trait SparkMemoryManager { + /// Asks Spark for `size` bytes and returns how many it granted. + fn acquire(&self, size: usize) -> CometResult<i64>; + fn release(&self, size: usize) -> CometResult<()>; +} + +/// Calls [`crate::jvm_bridge::CometTaskMemoryManager`] over JNI. +pub(super) struct JniMemoryManager(Arc<Global<JObject<'static>>>); + +impl SparkMemoryManager for JniMemoryManager { + fn acquire(&self, size: usize) -> CometResult<i64> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, + comet_task_memory_manager(handle).acquire_memory(size as i64) -> i64) + }) + } + + fn release(&self, size: usize) -> CometResult<()> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, comet_task_memory_manager(handle).release_memory(size as i64) -> ()) + }) + } +} + +/// Memory a Comet pool holds from Spark, including any it has recorded without Spark's grant. +/// +/// `MemoryPool::grow` must always succeed: DataFusion calls it for memory that already exists, +/// 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. +/// The pool's own `used` still counts the full amount, so its next `try_grow` is refused. +pub(super) struct SparkMemory<M = JniMemoryManager> { + manager: M, + overcommit: AtomicUsize, + task_attempt_id: i64, +} Review Comment: Two things about the shape here. `task_attempt_id` now lives in both `CometUnifiedMemoryPool` and its `SparkMemory`, and each only uses its own copy. Worth keeping one. The `<M = JniMemoryManager>` default generic exists purely so the tests can swap the manager out. A `Box<dyn SparkMemoryManager>` field would cost one virtual call next to a JNI call, drop the generic from both pools' field types, and make it straightforward to give `CometFairMemoryPool` and `CometUnifiedMemoryPool` a `#[cfg(test)]` constructor. That last part matters for the test comment below. ########## native/core/src/execution/memory_pools/fair_pool.rs: ########## @@ -51,6 +51,7 @@ impl Debug for CometFairMemoryPool { .field("pool_size", &self.pool_size) .field("used", &state.used) .field("num", &state.num) + .field("overcommit", &self.spark.overcommit()) Review Comment: `overcommit` is reachable only through `Debug`, and nothing in Comet formats these pools with `{:?}`, so the number never reaches a log line or an error message. The `Display` impls are what surface through the pool stack, and the `resources_err!` strings in `try_grow` are what an operator actually reads. Could it go in one of those instead? After this change, someone looking at an OOM-killed executor has no way to see how much unbacked memory the pool is carrying, and that is the number they need. The same applies to the `Debug` field added in `unified_pool.rs`. ########## native/core/src/execution/memory_pools/fair_pool.rs: ########## @@ -118,8 +105,18 @@ impl MemoryPool for CometFairMemoryPool { .expect("unexpected amount of unregister happened"); } + /// Records memory that already exists, so it must not fail and ignores the fair limit. + /// See [`SparkMemory`]. fn grow(&self, _reservation: &MemoryReservation, additional: usize) { - self.try_grow(_reservation, additional).unwrap(); + if additional == 0 { + return; + } + let mut state = self.state.lock(); + self.spark.acquire(additional); + state.used = state + .used + .checked_add(additional) + .expect("overflow in checked_add"); Review Comment: The two pools disagree on overflow now. This one panics through `.expect("overflow in checked_add")`, while `unified_pool.rs:101` uses `fetch_add` and wraps. `grow` is the one method that is contractually not allowed to fail, so a panic here works against what the PR is doing, and a silent wrap on the other side is not better. `saturating_add` in both would make them agree and keep `grow` infallible. ########## native/core/src/execution/memory_pools/spark_memory.rs: ########## @@ -0,0 +1,223 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::{ + atomic::{AtomicUsize, Ordering::Relaxed}, + Arc, +}; + +use jni::objects::{Global, JObject}; +use log::warn; + +use crate::{errors::CometResult, jvm_bridge::JVMClasses}; + +/// Spark's side of a Comet pool: the calls that acquire and release off-heap execution memory. +pub(super) trait SparkMemoryManager { + /// Asks Spark for `size` bytes and returns how many it granted. + fn acquire(&self, size: usize) -> CometResult<i64>; + fn release(&self, size: usize) -> CometResult<()>; +} + +/// Calls [`crate::jvm_bridge::CometTaskMemoryManager`] over JNI. +pub(super) struct JniMemoryManager(Arc<Global<JObject<'static>>>); + +impl SparkMemoryManager for JniMemoryManager { + fn acquire(&self, size: usize) -> CometResult<i64> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, + comet_task_memory_manager(handle).acquire_memory(size as i64) -> i64) + }) + } + + fn release(&self, size: usize) -> CometResult<()> { + let handle = self.0.as_obj(); + JVMClasses::with_env(|env| unsafe { + jni_call!(env, comet_task_memory_manager(handle).release_memory(size as i64) -> ()) + }) + } +} + +/// Memory a Comet pool holds from Spark, including any it has recorded without Spark's grant. +/// +/// `MemoryPool::grow` must always succeed: DataFusion calls it for memory that already exists, +/// 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. +/// The pool's own `used` still counts the full amount, so its next `try_grow` is refused. +pub(super) struct SparkMemory<M = JniMemoryManager> { + manager: M, + overcommit: AtomicUsize, + task_attempt_id: i64, +} + +impl SparkMemory { + pub(super) fn new(handle: Arc<Global<JObject<'static>>>, task_attempt_id: i64) -> Self { + Self::with_manager(JniMemoryManager(handle), task_attempt_id) + } +} + +impl<M: SparkMemoryManager> SparkMemory<M> { + fn with_manager(manager: M, task_attempt_id: i64) -> Self { + Self { + manager, + overcommit: AtomicUsize::new(0), + task_attempt_id, + } + } + + /// Acquires `size` bytes, or none: a partial grant is handed back and reported as `Err` with + /// the number of bytes Spark offered. + pub(super) fn try_acquire(&self, size: usize) -> CometResult<Result<(), usize>> { + let granted = granted(size, self.manager.acquire(size)?); + if granted < size { + self.manager.release(granted)?; + return Ok(Err(granted)); + } + Ok(Ok(())) + } + + /// Acquires what Spark will grant toward `size` bytes and carries the rest as overcommit. + /// Never fails; a failed call to Spark counts as a zero grant. + pub(super) fn acquire(&self, size: usize) { + let granted = match self.manager.acquire(size) { + Ok(acquired) => granted(size, acquired), + Err(e) => { + warn!( + "Task {} failed to acquire {size} bytes from Spark: {e:?}", + self.task_attempt_id + ); + 0 + } + }; + if granted < size { + self.overcommit.fetch_add(size - granted, Relaxed); + } + } + + /// Frees `size` bytes, repaying overcommit before releasing the rest to Spark. + pub(super) fn release(&self, size: usize) -> CometResult<()> { + let mut to_release = size; + if self.overcommit.load(Relaxed) > 0 { + let prev = self + .overcommit + .fetch_update(Relaxed, Relaxed, |debt| Some(debt.saturating_sub(size))) + .unwrap(); + to_release = size.saturating_sub(prev); + } + if to_release > 0 { + self.manager.release(to_release)?; + } + Ok(()) + } + + pub(super) fn overcommit(&self) -> usize { + self.overcommit.load(Relaxed) + } +} + +/// Clamps Spark's reply to an acquire: it never grants more than asked, and never a negative. +fn granted(requested: usize, acquired: i64) -> usize { + usize::try_from(acquired).unwrap_or(0).min(requested) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::errors::CometError; + use std::sync::Mutex; + + /// Grants at most `available` bytes and records every release. + #[derive(Default)] + struct FakeSpark { + available: Mutex<i64>, + released: Mutex<Vec<usize>>, + fail: bool, + } + + impl FakeSpark { + fn with(available: i64) -> Self { + Self { + available: Mutex::new(available), + ..Default::default() + } + } + } + + impl SparkMemoryManager for FakeSpark { + fn acquire(&self, size: usize) -> CometResult<i64> { + if self.fail { + return Err(CometError::Internal("jni".to_string())); + } + let mut available = self.available.lock().unwrap(); + let granted = (size as i64).min(*available); + *available -= granted; + Ok(granted) + } + + fn release(&self, size: usize) -> CometResult<()> { + *self.available.lock().unwrap() += size as i64; + self.released.lock().unwrap().push(size); + Ok(()) + } + } + + #[test] + fn try_acquire_hands_back_a_partial_grant() { + let spark = SparkMemory::with_manager(FakeSpark::with(40), 0); + assert_eq!(spark.try_acquire(100).unwrap(), Err(40)); + assert_eq!(*spark.manager.released.lock().unwrap(), vec![40]); + assert_eq!(spark.try_acquire(40).unwrap(), Ok(())); + assert_eq!(spark.overcommit(), 0); + } + + #[test] + fn release_repays_overcommit_before_returning_bytes_to_spark() { + let spark = SparkMemory::with_manager(FakeSpark::with(40), 0); + spark.acquire(100); + assert_eq!(spark.overcommit(), 60); + spark.release(30).unwrap(); + assert_eq!(spark.overcommit(), 30); + assert!(spark.manager.released.lock().unwrap().is_empty()); + spark.release(70).unwrap(); + assert_eq!(spark.overcommit(), 0); + // Spark gets back exactly the 40 bytes it granted. + assert_eq!(*spark.manager.released.lock().unwrap(), vec![40]); + } + + #[test] + fn failed_acquire_is_all_overcommit() { + let spark = SparkMemory::with_manager( + FakeSpark { + fail: true, + ..Default::default() + }, + 0, + ); + spark.acquire(64); + assert_eq!(spark.overcommit(), 64); + spark.release(64).unwrap(); + assert!(spark.manager.released.lock().unwrap().is_empty()); + } + + #[test] + fn granted_is_clamped_to_the_request() { + assert_eq!(granted(100, 40), 40); + assert_eq!(granted(100, 150), 100); + assert_eq!(granted(100, -1), 0); + } Review Comment: These four all still pass with `fn grow(..) { self.try_grow(..).unwrap() }` put back, so none of them distinguishes the fix from the bug. They are good tests of `SparkMemory`, but the behavior change is in the pools. With a test constructor on the pools, as suggested above, you could pin what the PR is actually claiming. A `grow` that exceeds `pool_size / num` succeeds and lands in `used`. The `try_grow` right after it is refused. A later `shrink` hands Spark back exactly what it granted and no more. That would also settle the `greedy_unified` question from my first comment one way or the other. -- 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]
