| 170 | } |
| 171 | |
| 172 | bool ParallelReadBuffer::nextImpl() |
| 173 | { |
| 174 | while (true) |
| 175 | { |
| 176 | /// All readers processed, stop |
| 177 | if (read_workers.empty()) |
| 178 | { |
| 179 | chassert(next_range_start >= file_size); |
| 180 | return false; |
| 181 | } |
| 182 | |
| 183 | auto * w = read_workers.front().get(); |
| 184 | |
| 185 | std::unique_lock lock{w->worker_mutex}; |
| 186 | |
| 187 | if (emergency_stop) |
| 188 | handleEmergencyStop(); // throws |
| 189 | |
| 190 | /// Read data from front reader |
| 191 | if (w->bytes_produced > w->bytes_consumed) |
| 192 | { |
| 193 | chassert(w->start_offset + w->bytes_consumed == static_cast<size_t>(current_position)); |
| 194 | |
| 195 | working_buffer = internal_buffer = Buffer( |
| 196 | w->segment.data() + w->bytes_consumed, w->segment.data() + w->bytes_produced); |
| 197 | current_position += working_buffer.size(); |
| 198 | w->bytes_consumed = w->bytes_produced; |
| 199 | |
| 200 | return true; |
| 201 | } |
| 202 | |
| 203 | /// Front reader is done, remove it and add another |
| 204 | if (!w->hasBytesToProduce()) |
| 205 | { |
| 206 | lock.unlock(); |
| 207 | read_workers.pop_front(); |
| 208 | addReaders(); |
| 209 | |
| 210 | continue; |
| 211 | } |
| 212 | |
| 213 | /// Nothing to do right now, wait for something to change. |
| 214 | /// |
| 215 | /// The timeout is a workaround for a race condition. |
| 216 | /// emergency_stop is assigned while holding a *different* mutex from the one we're holding |
| 217 | /// (exception_mutex vs worker_mutex). So it's possible that our emergency_stop check (above) |
| 218 | /// happens before a onBackgroundException() call, but our wait(lock) happens after it. |
| 219 | /// Then the wait may get stuck forever. |
| 220 | /// |
| 221 | /// Note that using wait(lock, [&]{ return emergency_stop || ...; }) wouldn't help because |
| 222 | /// it does effectively the same "check, then wait" sequence. |
| 223 | /// |
| 224 | /// One possible proper fix would be to make onBackgroundException() lock all read_workers |
| 225 | /// mutexes too (not necessarily simultaneously - just locking+unlocking them one by one |
| 226 | /// between the emergency_stop change and the notify_all() would be enough), but then we |
| 227 | /// need another mutex to protect read_workers itself... |
| 228 | next_condvar.wait_for(lock, std::chrono::seconds(10)); |
| 229 | } |