This is an automated email from the ASF dual-hosted git repository.
milenkovicm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new f75ec91b fix: executor_shutdown_while_running test has race (#1248)
f75ec91b is described below
commit f75ec91b4e1385213329809be84787a3f7fbeab5
Author: Marko Milenković <[email protected]>
AuthorDate: Tue Apr 22 10:47:14 2025 +0100
fix: executor_shutdown_while_running test has race (#1248)
leading for barrier never to get triggered
---
ballista/executor/src/cpu_bound_executor.rs | 29 +++++++++++++++++++++++------
1 file changed, 23 insertions(+), 6 deletions(-)
diff --git a/ballista/executor/src/cpu_bound_executor.rs
b/ballista/executor/src/cpu_bound_executor.rs
index d6c53bc4..5f1ed229 100644
--- a/ballista/executor/src/cpu_bound_executor.rs
+++ b/ballista/executor/src/cpu_bound_executor.rs
@@ -311,17 +311,23 @@ mod tests {
}
#[tokio::test]
- #[ignore]
- // related https://github.com/apache/arrow-datafusion/issues/2140
async fn executor_shutdown_while_task_running() {
- let barrier = Arc::new(Barrier::new(2));
+ let barrier_task_completed = Arc::new(Barrier::new(2));
+ // prevents executor shutdown before task gets running
+ // otherwise there is race on task spawn and shutdown
+ // leading barrier_task_completed waiting forever
+ let barrier_task_running = Arc::new(Barrier::new(2));
let exec = DedicatedExecutor::new("Test DedicatedExecutor", 1);
- let dedicated_task = exec.spawn(do_work(42, Arc::clone(&barrier)));
-
+ let dedicated_task = exec.spawn(signal_running_do_work(
+ 42,
+ Arc::clone(&barrier_task_running),
+ Arc::clone(&barrier_task_completed),
+ ));
+ barrier_task_running.wait();
exec.shutdown();
// block main thread until completion of the outstanding task
- barrier.wait();
+ barrier_task_completed.wait();
// task should complete successfully
assert_eq!(dedicated_task.await.unwrap(), 42);
@@ -375,4 +381,15 @@ mod tests {
barrier.wait();
result
}
+ /// signals when task starts running,
+ /// waits on barrier and then returns result
+ async fn signal_running_do_work(
+ result: usize,
+ barrier_task_running: Arc<Barrier>,
+ barrier_task_finished: Arc<Barrier>,
+ ) -> usize {
+ barrier_task_running.wait();
+ barrier_task_finished.wait();
+ result
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
