| @@ -36,7 +36,7 @@ mod serial { · 35 unmodified lines |
| 36 | 36 | |
| 37 | 37 | #[cfg(feature = "parallel")] |
| 38 | 38 | mod in_parallel { |
| 39 | | − use crate::parallel::Reducer; |
| 39 | + use crate::parallel::{num_threads, Reducer}; |
| 40 | 40 | use crossbeam_utils::thread; |
| 41 | 41 | |
| 42 | 42 | pub fn join<O1: Send, O2: Send>(left: impl FnOnce() -> O1 + Send, right: impl FnOnce() -> O2 + Send) -> (O1, O2) { |
| @@ -48,13 +48,6 @@ mod in_parallel { · 5 unmodified lines |
| 48 | 48 | .unwrap() |
| 49 | 49 | } |
| 50 | 50 | |
| 51 | | − fn num_threads(thread_limit: Option<usize>) -> usize { |
| 52 | | − let logical_cores = num_cpus::get(); |
| 53 | | − thread_limit |
| 54 | | − .map(|l| if l == 0 { logical_cores } else { l }) |
| 55 | | − .unwrap_or(logical_cores) |
| 56 | | − } |
| 57 | | − |
| 58 | 51 | pub fn in_parallel<I, S, O, R>( |
| 59 | 52 | input: impl Iterator<Item = I> + Send, |
| 60 | 53 | thread_limit: Option<usize>, |
| @@ -107,16 +100,6 @@ mod in_parallel { · 46 unmodified lines |
| 107 | 100 | } |
| 108 | 101 | } |
| 109 | 102 | |
| 110 | | −pub fn num_threads(thread_limit: Option<usize>) -> usize { |
| 111 | | − #[cfg(not(feature = "num_cpus"))] |
| 112 | | − return 1; |
| 113 | | − |
| 114 | | − let logical_cores = num_cpus::get(); |
| 115 | | − thread_limit |
| 116 | | − .map(|l| if l == 0 { logical_cores } else { l }) |
| 117 | | − .unwrap_or(logical_cores) |
| 118 | | −} |
| 119 | | − |
| 120 | 103 | pub fn optimize_chunk_size_and_thread_limit( |
| 121 | 104 | desired_chunk_size: usize, |
| 122 | 105 | num_chunks: Option<usize>, |
| @@ -144,7 +127,19 @@ pub fn optimize_chunk_size_and_thread_limit( · 21 unmodified lines |
| 144 | 127 | }; |
| 145 | 128 | (chunk_size, thread_limit) |
| 146 | 129 | }) |
| 147 | | − .unwrap_or((desired_chunk_size, available_threads)); |
| 130 | + .unwrap_or(( |
| 131 | + if available_threads == 1 { |
| 132 | + desired_chunk_size |
| 133 | + } else { |
| 134 | + let arbitrary_desirable_chunk_size = 5; |
| 135 | + if desired_chunk_size < arbitrary_desirable_chunk_size { |
| 136 | + arbitrary_desirable_chunk_size |
| 137 | + } else { |
| 138 | + desired_chunk_size |
| 139 | + } |
| 140 | + }, |
| 141 | + available_threads, |
| 142 | + )); |
| 148 | 143 | (chunk_size, Some(thread_limit)) |
| 149 | 144 | } |
| 150 | 145 | |
| @@ -154,6 +149,16 @@ pub use serial::*; · 3 unmodified lines |
| 154 | 149 | #[cfg(feature = "parallel")] |
| 155 | 150 | pub use in_parallel::*; |
| 156 | 151 | |
| 152 | +pub(crate) fn num_threads(thread_limit: Option<usize>) -> usize { |
| 153 | + #[cfg(not(feature = "num_cpus"))] |
| 154 | + return 1; |
| 155 | + |
| 156 | + let logical_cores = num_cpus::get(); |
| 157 | + thread_limit |
| 158 | + .map(|l| if l == 0 { logical_cores } else { l }) |
| 159 | + .unwrap_or(logical_cores) |
| 160 | +} |
| 161 | + |
| 157 | 162 | pub fn in_parallel_if<I, S, O, R>( |
| 158 | 163 | condition: impl FnOnce() -> bool, |
| 159 | 164 | input: impl Iterator<Item = I> + Send, |
| @@ -167,7 +172,7 @@ where · 7 unmodified lines |
| 167 | 172 | I: Send, |
| 168 | 173 | O: Send, |
| 169 | 174 | { |
| 170 | | − if condition() { |
| 175 | + if num_threads(thread_limit) > 1 && condition() { |
| 171 | 176 | in_parallel(input, thread_limit, new_thread_state, consume, reducer) |
| 172 | 177 | } else { |
| 173 | 178 | serial::in_parallel(input, thread_limit, new_thread_state, consume, reducer) |