[arrow-datafusion] branch master updated: Add debug log when waiting for spilling on other consumers (#1933)

[houqp]

This is an automated email from the ASF dual-hosted git repository.

houqp pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git


The following commit(s) were added to refs/heads/master by this push:
     new 741eba0  Add debug log when waiting for spilling on other consumers 
(#1933)
741eba0 is described below

commit 741eba08423b7a14473958ab2067e125e706e74c
Author: Liang-Chi Hsieh <vii...@gmail.com>
AuthorDate: Mon Mar 7 23:17:02 2022 -0800

    Add debug log when waiting for spilling on other consumers (#1933)
    
    * Add debug log when waiting
    
    * Measure waiting time
---
 datafusion/src/execution/memory_manager.rs | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/datafusion/src/execution/memory_manager.rs 
b/datafusion/src/execution/memory_manager.rs
index d39eaab..e48585e 100644
--- a/datafusion/src/execution/memory_manager.rs
+++ b/datafusion/src/execution/memory_manager.rs
@@ -20,12 +20,13 @@
 use crate::error::{DataFusionError, Result};
 use async_trait::async_trait;
 use hashbrown::HashSet;
-use log::debug;
+use log::{debug, warn};
 use parking_lot::{Condvar, Mutex};
 use std::fmt;
 use std::fmt::{Debug, Display, Formatter};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::time::{Duration, Instant};
 
 static CONSUMER_ID: AtomicUsize = AtomicUsize::new(0);
 
@@ -340,7 +341,15 @@ impl MemoryManager {
             } else if current < min_per_rqt {
                 // if we cannot acquire at lease 1/2n memory, just wait for 
others
                 // to spill instead spill self frequently with limited total 
mem
+                debug!(
+                    "Cannot acquire minimum amount of memory {} on memory 
manager {}, waiting for others to spill ...",
+                    human_readable_size(min_per_rqt), self);
+                let now = Instant::now();
                 self.cv.wait(&mut rqt_current_used);
+                let elapsed = now.elapsed();
+                if elapsed > Duration::from_secs(10) {
+                    warn!("Elapsed on waiting for spilling: {:.2?}", elapsed);
+                }
             } else {
                 granted = false;
                 break;