This functionality is similar to parallelFor, except that reasoning about the number of shards used is significantly easier.
| 148 | // This functionality is similar to parallelFor, except that reasoning about |
| 149 | // the number of shards used is significantly easier. |
| 150 | void ThreadPool::TransformRangeConcurrently( |
| 151 | const int64 block_size, const int64 total, |
| 152 | const std::function<void(int64, int64)>& fn) { |
| 153 | const int num_shards_used = |
| 154 | NumShardsUsedByTransformRangeConcurrently(block_size, total); |
| 155 | if (num_shards_used == 1) { |
| 156 | fn(0, total); |
| 157 | return; |
| 158 | } |
| 159 | |
| 160 | // Adapted from Eigen's parallelFor implementation. |
| 161 | BlockingCounter counter(num_shards_used); |
| 162 | std::function<void(int64, int64)> handle_range = |
| 163 | [=, &handle_range, &counter, &fn](int64 first, int64 last) { |
| 164 | while (last - first > block_size) { |
| 165 | // Find something near the midpoint which is a multiple of block size. |
| 166 | const int64 mid = first + ((last - first) / 2 + block_size - 1) / |
| 167 | block_size * block_size; |
| 168 | Schedule([=, &handle_range]() { handle_range(mid, last); }); |
| 169 | last = mid; |
| 170 | } |
| 171 | // Single block or less, execute directly. |
| 172 | fn(first, last); |
| 173 | counter.DecrementCount(); // The shard is done. |
| 174 | }; |
| 175 | if (num_shards_used <= NumThreads()) { |
| 176 | // Avoid a thread hop by running the root of the tree and one block on the |
| 177 | // main thread. |
| 178 | handle_range(0, total); |
| 179 | } else { |
| 180 | // Execute the root in the thread pool to avoid running work on more than |
| 181 | // numThreads() threads. |
| 182 | Schedule([=, &handle_range]() { handle_range(0, total); }); |
| 183 | } |
| 184 | counter.Wait(); |
| 185 | } |
| 186 | |
| 187 | void ThreadPool::ParallelFor(int64 total, int64 cost_per_unit, |
| 188 | std::function<void(int64, int64)> fn) { |