You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "wiedld (via GitHub)" <gi...@apache.org> on 2024/07/26 02:36:05 UTC
[PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
wiedld opened a new pull request, #11665:
URL: https://github.com/apache/datafusion/pull/11665
## Which issue does this PR close?
Part of #11523
## Rationale for this change
The OOMing error message returns only the errors which occurred for the next incremental request of memory, and not for the biggest consumers of memory. As a result, we have spent time chasing OOMs in the wrong place.
## What changes are included in this PR?
Includes a new MemoryPool implementation, which when (optionally) used with an inner Memorypool will return a list of the top memory consumers on err.
This new `TrackConsumersPool` is optional, such that individual users can determine whether or not to use this capability (along with any associated overhead).
## Are these changes tested?
Yes.
## Are there any user-facing changes?
Yes, as a new `TrackConsumersPool`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#issuecomment-2256227324
@alamb -- I cannot reproduce these test failures locally, even with the same flags & tokio multithreaded tests enabled. Can you help?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` that has better error messages on exhaustion [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1697466468
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +463,190 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5";
+ let res = r5.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers, instead found {:?}",
+ res
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100";
+ let res = r0.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet, instead found {:?}", res
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 10 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90";
+ let res = r1.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90), instead found {:?}", res
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70";
Review Comment:
Please feel free to suggest better wording for this^^. 🙏🏼
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#issuecomment-2256389441
> @alamb -- I cannot reproduce these test failures locally, even with the same flags & tokio multithreaded tests enabled. Can you help?
I also could not reproduce it with
```
cargo test --lib -p datafusion-execution
```
One way to debug such failures is to add additional debug logging to the tests -- right now the test failure simply says "failed" with a static message, but it doesn't say what the actual result was
For example you could potentially update your asserts like this:
```diff
diff --git a/datafusion/execution/src/memory_pool/pool.rs b/datafusion/execution/src/memory_pool/pool.rs
index 9ce14e41a..0df795a2b 100644
--- a/datafusion/execution/src/memory_pool/pool.rs
+++ b/datafusion/execution/src/memory_pool/pool.rs
@@ -494,12 +494,13 @@ mod tests {
// using the previously set sizes for other consumers
let mut r5 = MemoryConsumer::new("r5").register(&pool);
let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ let actual_result = r5.try_grow(150);
assert!(
matches!(
- r5.try_grow(150),
+ actual_result,
Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
),
- "should provide list of top memory consumers"
+ "should provide list of top memory consumers, got {actual_result:?}",
);
}
```
And the message from your assertion failure should be more helpful for debugging
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#issuecomment-2256764400
> given the tests pass on amd64 maybe the difference is related to newlines or something in the message?
We have a backtrace inserted into the error message. Specifically, the expected is:
`Failed to allocate additional 150 bytes for r0 with 10 bytes already allocated - maximum available is 70. The top memory consumers (across reservations) are: r1 consumed 20 bytes, r0 consumed 10 bytes`
When running locally, we get that^^ response.
Yet in CI, we get the backtrace:
```
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692421398
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should error when no other consumers are reported"
+ );
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previous set size for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1693465283) in a follow PR, the final error message would read:
~`Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated for this reservation - 5 bytes remain available for the total pool. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes`~
`Resources exhausted with top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated for this reservation - 5 bytes remain available for the total pool`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692420852
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1692419610) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated for this reservation - 100 bytes remain available for the total pool. The top memory consumers (across reservations) are: r0 consumed 0 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692420852
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1692419610) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated for this reservation - only 100 bytes remain available for the total pool. The top memory consumers (across reservations) are: r0 consumed 0 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692420852
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1692419610) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated for this reservation - maximum available is 100 for the total pool. The top memory consumers (across reservations) are: r0 consumed 0 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693467375
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1693465283) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated for this reservation - 100 bytes remain available for the total pool. The top memory consumers (across reservations) are: r0 consumed 0 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#issuecomment-2256916459
[CI was failing due to the concat of error messages](https://github.com/apache/datafusion/pull/11665#issuecomment-2256764400). I fixed it two ways -- in order to demonstrate why the 2nd way is better.
* 1st way is to [splice the error message,](https://github.com/apache/datafusion/pull/11665/commits/09b20d289f53d3b61b976313f8731e8a6711f370) in order to insert the consumer. This inherently violates the nature of error bubbling IMO.
* 2nd way is to [wrap the OOM error with the top consumers](https://github.com/apache/datafusion/pull/11665/commits/f75764e9de36c4d336a6f84203c95a460a9d3610). IMO this is correct.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` that has better error messages on exhaustion [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1697466026
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +463,190 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5";
+ let res = r5.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers, instead found {:?}",
+ res
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100";
+ let res = r0.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet, instead found {:?}", res
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 10 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90";
+ let res = r1.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90), instead found {:?}", res
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70";
Review Comment:
I agree this is confusing. Because the error message `for foo with 20 bytes already allocated` is [the error from the MemoryReservation](https://github.com/apache/datafusion/blob/c3ce60f211c64e083897703adc076807c20e3617/datafusion/execution/src/memory_pool/pool.rs#L238-L244), which has 20 bytes.
Whereas the MemoryConsumer happens to have 2 reservations -- with a total of 30 bytes allotted (across reservations).
I'm hoping the followup PR (with the message change for the reservation-specific error) will make this more clear. Does the below read better?
`Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated for this reservation - 5 bytes remain available for the total pool
`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` with better error messages on exhaustion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #11665:
URL: https://github.com/apache/datafusion/pull/11665
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693521876
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet"
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90. The top memory consumers (across reservations) are: foo consumed 10 bytes";
+ assert!(
+ matches!(
+ r1.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90)"
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70. The top memory consumers (across reservations) are: foo consumed 30 bytes";
+ assert!(
+ matches!(
+ r1.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=30 bytes, available=70)"
+ );
+
+ // Test: different hashed consumer, (even with the same name),
+ // will be recognized as different in the TrackConsumersPool
+ let consumer_with_same_name_but_different_hash =
+ MemoryConsumer::new(same_name).with_can_spill(true);
+ let mut r2 = consumer_with_same_name_but_different_hash.register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 70. The top memory consumers (across reservations) are: foo(can_spill=false) consumed 30 bytes, foo(can_spill=true) consumed 0 bytes";
+ assert!(
+ matches!(
+ r2.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with different hashed consumer (foo(can_spill=false)=30 bytes and foo(can_spill=true)=0 bytes, available=70)"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_deregister() {
Review Comment:
This test is about an existing behavior. The current MemoryPool implementations will register/deregister on the MemoryConsumer level; whereas the bytes incr/decrement is on the MemoryReservation level.
This means that a consumer can be deregistered, even while the reservation still holds memory.
The new TrackConsumersPool has nothing to do with this existing behavior. However, this test is added in order to demonstrate what/how the error messages will read.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693521876
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet"
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90. The top memory consumers (across reservations) are: foo consumed 10 bytes";
+ assert!(
+ matches!(
+ r1.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90)"
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70. The top memory consumers (across reservations) are: foo consumed 30 bytes";
+ assert!(
+ matches!(
+ r1.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=30 bytes, available=70)"
+ );
+
+ // Test: different hashed consumer, (even with the same name),
+ // will be recognized as different in the TrackConsumersPool
+ let consumer_with_same_name_but_different_hash =
+ MemoryConsumer::new(same_name).with_can_spill(true);
+ let mut r2 = consumer_with_same_name_but_different_hash.register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 70. The top memory consumers (across reservations) are: foo(can_spill=false) consumed 30 bytes, foo(can_spill=true) consumed 0 bytes";
+ assert!(
+ matches!(
+ r2.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with different hashed consumer (foo(can_spill=false)=30 bytes and foo(can_spill=true)=0 bytes, available=70)"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_deregister() {
Review Comment:
This test is about an existing behavior. The current MemoryPool implementations will register/deregister on the MemoryConsumer level; whereas the bytes incr/decrement is on the MemoryReservation level.
This means that a consumer can be deregistered, even while the reservation still holds memory.
The new TrackConsumersPool has nothing to do with this existing behavior. However, this test is added in order to explain what/how the error messages will read.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693480102
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,11 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the `bytes already allocated` for the reservation,
+/// and not the total within the collective [`MemoryPool`],
+/// nor the total across multiple reservations with the same [`MemoryConsumer`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
This^^ suggestion is because this error message is inherently about only the reservation itself. Each reservation can request bytes from the pool -- but the pool itself only incr/decrements totals for the pool. It does **NOT** track each reservation, nor if multiple reservations have the same consumer.
As a result, the simplest solution is to: (a) update this error message, and (b) continue to rely upon each reservation to resize itself to zero upon drop ([as it does](https://github.com/apache/datafusion/blob/01dc3f9766e1e36ef9864effdb1a7c89d138a982/datafusion/execution/src/memory_pool/mod.rs#L307-L310)).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693480102
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,11 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the `bytes already allocated` for the reservation,
+/// and not the total within the collective [`MemoryPool`],
+/// nor the total across multiple reservations with the same [`MemoryConsumer`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
This^^ suggestion is because this error message is inherently about only the reservation itself. Each reservation can request bytes from the pool -- but the pool itself only incr/decrements totals for the pool. It does **NOT** track each reservation, nor if multiple reservations have the same consumer.
As a result, the simplest solution is to: (a) update this error message, and (b) continue to rely upon each reservation to resize itself to upon drop ([as it does](https://github.com/apache/datafusion/blob/01dc3f9766e1e36ef9864effdb1a7c89d138a982/datafusion/execution/src/memory_pool/mod.rs#L307-L310)).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` that has better error messages on exhaustion [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1697466026
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +463,190 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5";
+ let res = r5.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers, instead found {:?}",
+ res
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100";
+ let res = r0.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet, instead found {:?}", res
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 10 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90";
+ let res = r1.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90), instead found {:?}", res
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70";
Review Comment:
I agree this is confusing. Because the error message `for foo with 20 bytes already allocated` is [the error from the MemoryReservation](https://github.com/apache/datafusion/blob/c3ce60f211c64e083897703adc076807c20e3617/datafusion/execution/src/memory_pool/pool.rs#L238-L244), which has 20 bytes.
Whereas the MemoryConsumer happens to have 2 reservations -- with a total of 30 bytes allotted (across reservations).
I'm hoping the followup PR (with the message change for the reservation-specific error) will make this more clear. Does this seem more clear?
`Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated for this reservation - 5 bytes remain available for the total pool
`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` that has better error messages on exhaustion [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1697608547
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -240,6 +249,149 @@ fn insufficient_capacity_err(
resources_datafusion_err!("Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}", additional, reservation.registration.consumer.name, reservation.size, available)
}
+/// A [`MemoryPool`] that tracks the consumers that have
+/// reserved memory within the inner memory pool.
+///
+/// Tracking is per hashed [`MemoryConsumer`], not per [`MemoryReservation`].
+/// The same consumer can have multiple reservations.
+#[derive(Debug)]
+pub struct TrackConsumersPool<I> {
+ inner: I,
+ top: NonZeroUsize,
+ tracked_consumers: Mutex<HashMap<MemoryConsumer, AtomicU64>>,
+}
+
+impl<I: MemoryPool> TrackConsumersPool<I> {
+ /// Creates a new [`TrackConsumersPool`].
+ ///
+ /// The `top` determines how many Top K [`MemoryConsumer`]s to include
+ /// in the reported [`DataFusionError::ResourcesExhausted`].
+ pub fn new(inner: I, top: NonZeroUsize) -> Self {
+ Self {
+ inner,
+ top,
+ tracked_consumers: Default::default(),
+ }
+ }
+
+ /// Determine if there are multiple [`MemoryConsumer`]s registered
+ /// which have the same name.
+ ///
+ /// This is very tied to the implementation of the memory consumer.
+ fn has_multiple_consumers(&self, name: &String) -> bool {
+ let consumer = MemoryConsumer::new(name);
+ let consumer_with_spill = consumer.clone().with_can_spill(true);
+ let guard = self.tracked_consumers.lock();
+ guard.contains_key(&consumer) && guard.contains_key(&consumer_with_spill)
+ }
+
+ /// The top consumers in a report string.
+ fn report_top(&self) -> String {
Review Comment:
The use of TrackConsumersPool for error reporting (when passed as `Arc<dyn MemoryPool>`) is constrained by the trait definition. However, we could use the downcasted struct itself for runtime metrics [as shown in this added commit](https://github.com/apache/datafusion/pull/11665/commits/c8c01961dd54df1703d7efdf216a8f63db8656b4). Is this what you were thinking?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693467375
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1693465283) in a follow PR, the final error message would read:
~`Failed to allocate additional 150 bytes for foo with 0 bytes already allocated for this reservation - 100 bytes remain available for the total pool. The top memory consumers (across reservations) are: foo consumed 0 bytes`~
`Resources exhausted with top memory consumers (across reservations) are: foo consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated for this reservation - 100 bytes remain available for the total pool`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692421398
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should error when no other consumers are reported"
+ );
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previous set size for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1692419610) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated for this reservation - maximum available is 5 for the total pool. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693517734
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet"
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
Review Comment:
As noted in the struct docs, the TrackConsumersPool is all about the MemoryConsumer. Therefore the top K consumers are returned, even if a given consumer has multiple reservations. This is intended to make the error messages (and subsequent debugging) more useful.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692420852
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1692419610) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated for this reservation - 100 bytes remain available for the total pool. The top memory consumers (across reservations) are: r0 consumed 0 bytes`
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,10 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the total bytes affiliated with the reservation's
+/// [`MemoryConsumer`], and not the total within the collective [`MemoryPool`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
In a follow up PR, I would like to iterate on this error message.
Take the original/current:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}`
And change into something like:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated for this reservation - {} bytes remain available for the total pool`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692419610
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,10 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the total bytes affiliated with the reservation's
+/// [`MemoryConsumer`], and not the total within the collective [`MemoryPool`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
In a follow up PR, I would like to iterate on this error message.
Take the original/current:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}`
And change into something like:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated for this reservation - {} bytes remain available for the total pool`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1697242963
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -240,6 +249,149 @@ fn insufficient_capacity_err(
resources_datafusion_err!("Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}", additional, reservation.registration.consumer.name, reservation.size, available)
}
+/// A [`MemoryPool`] that tracks the consumers that have
+/// reserved memory within the inner memory pool.
+///
+/// Tracking is per hashed [`MemoryConsumer`], not per [`MemoryReservation`].
+/// The same consumer can have multiple reservations.
+#[derive(Debug)]
+pub struct TrackConsumersPool<I> {
+ inner: I,
+ top: NonZeroUsize,
+ tracked_consumers: Mutex<HashMap<MemoryConsumer, AtomicU64>>,
+}
+
+impl<I: MemoryPool> TrackConsumersPool<I> {
+ /// Creates a new [`TrackConsumersPool`].
+ ///
+ /// The `top` determines how many Top K [`MemoryConsumer`]s to include
+ /// in the reported [`DataFusionError::ResourcesExhausted`].
+ pub fn new(inner: I, top: NonZeroUsize) -> Self {
+ Self {
+ inner,
+ top,
+ tracked_consumers: Default::default(),
+ }
+ }
+
+ /// Determine if there are multiple [`MemoryConsumer`]s registered
+ /// which have the same name.
+ ///
+ /// This is very tied to the implementation of the memory consumer.
+ fn has_multiple_consumers(&self, name: &String) -> bool {
+ let consumer = MemoryConsumer::new(name);
+ let consumer_with_spill = consumer.clone().with_can_spill(true);
+ let guard = self.tracked_consumers.lock();
+ guard.contains_key(&consumer) && guard.contains_key(&consumer_with_spill)
+ }
+
+ /// The top consumers in a report string.
+ fn report_top(&self) -> String {
Review Comment:
since `top` is only used for this report maybe it should be a parameter to `report_top` rather than a parameter on the pool 🤔
It seems like someone could want both the "top 10" consumers and "all consumers" from the same tracked pool but with the implementation they could only have one or the other
Is this the message you are proposing to fix in a follow on?
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -240,6 +249,149 @@ fn insufficient_capacity_err(
resources_datafusion_err!("Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}", additional, reservation.registration.consumer.name, reservation.size, available)
}
+/// A [`MemoryPool`] that tracks the consumers that have
+/// reserved memory within the inner memory pool.
Review Comment:
Maybe we can also add some context on the usecase
```suggestion
/// reserved memory within the inner memory pool.
///
/// By tracking memory reservations more carefully this pool
/// can provide better error messages on the largest memory users
```
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,11 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the `bytes already allocated` for the reservation,
+/// and not the total within the collective [`MemoryPool`],
+/// nor the total across multiple reservations with the same [`MemoryConsumer`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
I like the idea of updating the error message in a follow on PR
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +463,190 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5";
+ let res = r5.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers, instead found {:?}",
+ res
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100";
+ let res = r0.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet, instead found {:?}", res
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 10 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90";
+ let res = r1.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90), instead found {:?}", res
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70";
Review Comment:
This is somewhat confusing that in the same message foo is reported to have two different allocations:
1. `foo consumed 30 bytes`
2. `for foo with 20 bytes already allocated`
Is it possible to rationalize the errors somehow make this less confusing?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#issuecomment-2256391575
given the tests pass on amd64 maybe the difference is related to newlines or something in the message?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` that has better error messages on exhaustion [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1697466026
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +463,190 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5";
+ let res = r5.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers, instead found {:?}",
+ res
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100";
+ let res = r0.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet, instead found {:?}", res
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 10 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90";
+ let res = r1.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90), instead found {:?}", res
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70";
Review Comment:
I agree this is confusing. Because the error message `for foo with 20 bytes already allocated` is [the error from the MemoryReservation](https://github.com/apache/datafusion/blob/c3ce60f211c64e083897703adc076807c20e3617/datafusion/execution/src/memory_pool/pool.rs#L238-L244), which has 20 bytes.
Whereas the MemoryConsumer happens to have 2 reservations -- with a total of 30 bytes allotted (across reservations).
I'm hoping the followup PR (with the message change) will make this more clear. Does this seem more clear?
`Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated for this reservation - 5 bytes remain available for the total pool
`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693467375
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1693465283) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for foo with 0 bytes already allocated for this reservation - 100 bytes remain available for the total pool. The top memory consumers (across reservations) are: foo consumed 0 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693465283
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,11 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the `bytes already allocated` for the reservation,
+/// and not the total within the collective [`MemoryPool`],
+/// nor the total across multiple reservations with the same [`MemoryConsumer`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
In a follow up PR, I would like to iterate on this error message.
Take the original/current:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}`
And change into something like:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated for this reservation - {} bytes remain available for the total pool`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692419610
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,10 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the total bytes affiliated with the reservation's
+/// [`MemoryConsumer`], and not the total within the collective [`MemoryPool`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
In a follow up PR, I would like to iterate on this error message.
Take the original/current:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}`
And change into something like:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated for this reservation - only {} bytes remain available for the total pool`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693521876
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet"
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90. The top memory consumers (across reservations) are: foo consumed 10 bytes";
+ assert!(
+ matches!(
+ r1.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90)"
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70. The top memory consumers (across reservations) are: foo consumed 30 bytes";
+ assert!(
+ matches!(
+ r1.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=30 bytes, available=70)"
+ );
+
+ // Test: different hashed consumer, (even with the same name),
+ // will be recognized as different in the TrackConsumersPool
+ let consumer_with_same_name_but_different_hash =
+ MemoryConsumer::new(same_name).with_can_spill(true);
+ let mut r2 = consumer_with_same_name_but_different_hash.register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 70. The top memory consumers (across reservations) are: foo(can_spill=false) consumed 30 bytes, foo(can_spill=true) consumed 0 bytes";
+ assert!(
+ matches!(
+ r2.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with different hashed consumer (foo(can_spill=false)=30 bytes and foo(can_spill=true)=0 bytes, available=70)"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_deregister() {
Review Comment:
This test is about an existing contract for how `MemoryPool::deregister()`, and how total bytes are accounted. The current MemoryPool implementations will register/deregister on the MemoryConsumer level; whereas the bytes incr/decrement is on the MemoryReservation level.
This means that a consumer can be deregistered, even while the reservation still holds memory.
The new TrackConsumersPool has nothing to do with this existing behavior. However, this test is added in order to explain what/how the error messages will read.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhaustion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1693517734
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +458,179 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
+ assert!(
+ matches!(
+ r5.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers"
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: foo consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet"
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
Review Comment:
As noted in the struct docs, the TrackConsumersPool is all about the MemoryConsumer. Therefore the top K consumers are returned, even if a given consumer has multiple reservations. This is intended to make the error messages (and subsequent debugging) more useful by correctly reflecting the aggregated top K.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` that has better error messages on exhaustion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1698612285
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -240,6 +249,152 @@ fn insufficient_capacity_err(
resources_datafusion_err!("Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}", additional, reservation.registration.consumer.name, reservation.size, available)
}
+/// A [`MemoryPool`] that tracks the consumers that have
+/// reserved memory within the inner memory pool.
+///
+/// By tracking memory reservations more carefully this pool
+/// can provide better error messages on the largest memory users
+///
+/// Tracking is per hashed [`MemoryConsumer`], not per [`MemoryReservation`].
+/// The same consumer can have multiple reservations.
+#[derive(Debug)]
+pub struct TrackConsumersPool<I> {
+ inner: I,
+ top: NonZeroUsize,
+ tracked_consumers: Mutex<HashMap<MemoryConsumer, AtomicU64>>,
+}
+
+impl<I: MemoryPool> TrackConsumersPool<I> {
+ /// Creates a new [`TrackConsumersPool`].
+ ///
+ /// The `top` determines how many Top K [`MemoryConsumer`]s to include
Review Comment:
👍
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692419610
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -231,6 +235,10 @@ impl MemoryPool for FairSpillPool {
}
}
+/// Constructs a resources error based upon the individual [`MemoryReservation`].
+///
+/// The error references the total bytes affiliated with the reservation's
+/// [`MemoryConsumer`], and not the total within the collective [`MemoryPool`].
#[inline(always)]
fn insufficient_capacity_err(
Review Comment:
In a follow up PR, I would like to iterate on this error message.
Take the original/current:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated - maximum available is {}`
And change into something like:
`Failed to allocate additional {} bytes for {} with {} bytes already allocated for this reservation - maximum available is {} for the total pool`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692421398
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should error when no other consumers are reported"
+ );
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previous set size for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1693465283) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated for this reservation - 5 bytes remain available for the total pool. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Provide actionable error messaging due to resource exhuastion. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1692421398
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +423,56 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new("r0").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r0 with 0 bytes already allocated - maximum available is 100. The top memory consumers (across reservations) are: r0 consumed 0 bytes";
+ assert!(
+ matches!(
+ r0.try_grow(150),
+ Err(DataFusionError::ResourcesExhausted(e)) if e.to_string().contains(expected)
+ ),
+ "should error when no other consumers are reported"
+ );
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previous set size for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes";
Review Comment:
With the [proposed change](https://github.com/apache/datafusion/pull/11665/files#r1692419610) in a follow PR, the final error message would read:
`Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated for this reservation - 5 bytes remain available for the total pool. The top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add `TrackedMemoryPool` that has better error messages on exhaustion [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #11665:
URL: https://github.com/apache/datafusion/pull/11665#discussion_r1697466026
##########
datafusion/execution/src/memory_pool/pool.rs:
##########
@@ -311,4 +463,190 @@ mod tests {
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated - maximum available is 20");
}
+
+ #[test]
+ fn test_tracked_consumers_pool() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ // Test: use all the different interfaces to change reservation size
+
+ // set r1=50, using grow and shrink
+ let mut r1 = MemoryConsumer::new("r1").register(&pool);
+ r1.grow(70);
+ r1.shrink(20);
+
+ // set r2=15 using try_grow
+ let mut r2 = MemoryConsumer::new("r2").register(&pool);
+ r2.try_grow(15)
+ .expect("should succeed in memory allotment for r2");
+
+ // set r3=20 using try_resize
+ let mut r3 = MemoryConsumer::new("r3").register(&pool);
+ r3.try_resize(25)
+ .expect("should succeed in memory allotment for r3");
+ r3.try_resize(20)
+ .expect("should succeed in memory allotment for r3");
+
+ // set r4=10
+ // this should not be reported in top 3
+ let mut r4 = MemoryConsumer::new("r4").register(&pool);
+ r4.grow(10);
+
+ // Test: reports if new reservation causes error
+ // using the previously set sizes for other consumers
+ let mut r5 = MemoryConsumer::new("r5").register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: r1 consumed 50 bytes, r3 consumed 20 bytes, r2 consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated - maximum available is 5";
+ let res = r5.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide list of top memory consumers, instead found {:?}",
+ res
+ );
+ }
+
+ #[test]
+ fn test_tracked_consumers_pool_register() {
+ let pool: Arc<dyn MemoryPool> = Arc::new(TrackConsumersPool::new(
+ GreedyMemoryPool::new(100),
+ NonZeroUsize::new(3).unwrap(),
+ ));
+
+ let same_name = "foo";
+
+ // Test: see error message when no consumers recorded yet
+ let mut r0 = MemoryConsumer::new(same_name).register(&pool);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 100";
+ let res = r0.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error when no reservations have been made yet, instead found {:?}", res
+ );
+
+ // API: multiple registrations using the same hashed consumer,
+ // will be recognized as the same in the TrackConsumersPool.
+
+ // Test: will be the same per Top Consumers reported.
+ r0.grow(10); // make r0=10, pool available=90
+ let new_consumer_same_name = MemoryConsumer::new(same_name);
+ let mut r1 = new_consumer_same_name.clone().register(&pool);
+ // TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
+ // a followup PR will clarify this message "0 bytes already allocated for this reservation"
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 10 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated - maximum available is 90";
+ let res = r1.try_grow(150);
+ assert!(
+ matches!(
+ &res,
+ Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(expected)
+ ),
+ "should provide proper error with same hashed consumer (a single foo=10 bytes, available=90), instead found {:?}", res
+ );
+
+ // Test: will accumulate size changes per consumer, not per reservation
+ r1.grow(20);
+ let expected = "Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated - maximum available is 70";
Review Comment:
I agree this is confusing. Because the error message `for foo with 20 bytes already allocated` is [the error from the MemoryReservation](https://github.com/apache/datafusion/blob/c3ce60f211c64e083897703adc076807c20e3617/datafusion/execution/src/memory_pool/pool.rs#L238-L244), which has 20 bytes.
Whereas the MemoryConsumer happens to have 2 reservations -- with a total of 30 bytes allotted (across reservations).
I'm hoping the followup PR (with the message change for the reservation-specific error) will make this more clear. Does the below read better?
`Resources exhausted with top memory consumers (across reservations) are: foo consumed 30 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated for this reservation - 70 bytes remain available for the total pool
`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org