dwsmith1983 commented on code in PR #5613:
URL: https://github.com/apache/datafusion-comet/pull/5613#discussion_r4362172432
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -202,9 +278,354 @@ impl MemoryPool for CometFairMemoryPool {
#[cfg(test)]
mod tests {
- use super::super::spark_memory::fake::FakeSpark;
+ use super::super::spark_memory::{fake::FakeSpark, SparkMemoryManager};
use super::*;
- use std::sync::Arc;
+ use crate::errors::{CometError, CometResult};
+ use parking_lot::Condvar;
+ use std::collections::{hash_map::Entry, HashMap};
+ use std::sync::atomic::{AtomicBool, AtomicI64, AtomicUsize,
Ordering::SeqCst};
+ use std::sync::mpsc::{channel, Receiver, Sender};
+ use std::sync::{Arc, Barrier};
+ use std::thread;
+ use std::time::Duration;
+
+ const MIB: usize = 1 << 20;
+ const GIB: usize = 1 << 30;
+
+ const THIS_TASK: i64 = 0;
+ const OTHER_TASK: i64 = 1;
+ const THIRD_TASK: i64 = 2;
+
+ /// Real Spark parks a starved acquire forever; the stub panics after this
so a deadlock
+ /// fails the test instead of hanging it.
+ const WAIT_TIMEOUT: Duration = Duration::from_secs(5);
+ const TEST_TIMEOUT: Duration = Duration::from_secs(20);
+
+ /// While armed, holds a bridge call until the test opens it, to
interleave another thread.
+ struct Gate {
+ armed: AtomicBool,
+ entered: (Sender<()>, Mutex<Receiver<()>>),
+ open: (Sender<()>, Mutex<Receiver<()>>),
+ }
+
+ impl Gate {
+ fn new() -> Self {
+ let entered = channel();
+ let open = channel();
+ Self {
+ armed: AtomicBool::new(false),
+ entered: (entered.0, Mutex::new(entered.1)),
+ open: (open.0, Mutex::new(open.1)),
+ }
+ }
+
+ fn arm(&self) {
+ self.armed.store(true, SeqCst);
+ }
+
+ fn disarm(&self) {
+ self.armed.store(false, SeqCst);
+ }
+
+ fn pass(&self) {
+ if self.armed.load(SeqCst) {
+ let _ = self.entered.0.send(());
+ // A test that fails before opening the gate must not hang the
bridge thread.
+ let _ = self.open.1.lock().recv_timeout(TEST_TIMEOUT);
+ }
+ }
+
+ fn wait_entered(&self, what: &str) {
+ self.entered
+ .1
+ .lock()
+ .recv_timeout(TEST_TIMEOUT)
+ .unwrap_or_else(|_| panic!("{what} never reached the gate"));
+ }
+
+ fn open(&self) {
+ let _ = self.open.0.send(());
+ }
+ }
+
+ /// The state Spark's `ExecutionMemoryPool` keeps under its `lock` monitor.
+ struct SparkPool {
+ pool_size: i64,
+ /// `memoryForTask`: created on a task's first acquire, removed when a
release drains it.
+ memory_for_task: HashMap<i64, i64>,
+ }
+
+ impl SparkPool {
+ fn memory_free(&self) -> i64 {
+ self.pool_size - self.memory_for_task.values().sum::<i64>()
+ }
+ }
+
+ /// Models Spark 4.1.3's `ExecutionMemoryPool` acquire and release for
this task: the entry
+ /// lifecycle, the 1/N and 1/(2N) shares and the wait loop. Re-adding the
entry in the wait
+ /// loop stands in for the `CometTaskMemoryManager` retry, which these
tests do not cover.
+ struct StubTaskMemory {
Review Comment:
> Should the stub model the task monitor that `acquireExecutionMemory`
holds, for example a per-task mutex held across `acquire`?
Yes. `StubTaskMemory` has a `monitor` mutex now: `grant()` takes it first
and holds it through the wait loop, as `acquireExecutionMemory` does, and
`hand_back()` doesn't take it. The four tests you listed are gone, and the
stub's `requests`, `granted_gate`, `fail_release` and third task went with them.
--
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]
