LiaCastaneda commented on code in PR #26242:
URL: https://github.com/apache/datafusion/pull/26242#discussion_r4281779352
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -98,6 +102,38 @@ impl ExternalSorterMetrics {
}
}
+/// Limits concurrent batch sorting across the partitions of one [`SortExec`].
+#[derive(Debug, Clone)]
+struct SortConcurrency {
+ /// Each partition can use the full worker budget, but queues at most this
+ /// many jobs so it cannot put all its runs ahead of other partitions.
+ per_partition_limit: usize,
+ /// Shared budget: a job holds a permit until its run has been sorted.
+ permits: Arc<Semaphore>,
+}
+
+impl SortConcurrency {
+ /// Returns the shared limit on multithreaded Tokio runtimes, initialized
+ /// from the current runtime's worker count on first use.
+ fn for_current_runtime(cell: &OnceLock<Self>) -> Option<Self> {
+ let handle = Handle::try_current().ok()?;
+ if handle.runtime_flavor() != RuntimeFlavor::MultiThread {
+ return None;
+ }
+ Some(
+ cell.get_or_init(|| {
+ // Sorting is CPU-bound; more jobs would only queue for a
worker.
+ let workers = handle.metrics().num_workers();
Review Comment:
` num_workers()` is not necessarily the number of cores, right? Would it
make sense to use `get_available_parallelism()` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
