gabotechs commented on code in PR #16093:
URL: https://github.com/apache/datafusion/pull/16093#discussion_r2095419949
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -158,28 +225,17 @@ impl RepartitionExecState {
));
spawned_tasks.push(wait_for_task);
}
-
- Self {
+ *self = Self::ConsumingInputStreams(ConsumingInputStreamsState {
channels,
abort_helper: Arc::new(spawned_tasks),
+ });
+ match self {
+ RepartitionExecState::ConsumingInputStreams(value) => Ok(value),
+ _ => unreachable!(),
}
}
}
-/// Lazily initialized state
-///
-/// Note that the state is initialized ONCE for all partitions by a single
task(thread).
-/// This may take a short while. It is also like that multiple threads
-/// call execute at the same time, because we have just started "target
partitions" tasks
-/// which is commonly set to the number of CPU cores and all call execute at
the same time.
-///
-/// Thus, use a **tokio** `OnceCell` for this initialization so as not to
waste CPU cycles
-/// in a mutex lock but instead allow other threads to do something useful.
-///
-/// Uses a parking_lot `Mutex` to control other accesses as they are very
short duration
-/// (e.g. removing channels on completion) where the overhead of `await` is
not warranted.
-type LazyState = Arc<tokio::sync::OnceCell<Mutex<RepartitionExecState>>>;
Review Comment:
cc @crepererum as you are the original author of this approach
--
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-unsubscr...@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For additional commands, e-mail: github-h...@datafusion.apache.org
