| 257 | } |
| 258 | |
| 259 | void RuntimeState::Cancel() { |
| 260 | is_cancelled_.Store(true); |
| 261 | { |
| 262 | lock_guard<SpinLock> l(cancellation_cvs_lock_); |
| 263 | for (pair<std::mutex*, ConditionVariable*>& entry : cancellation_cvs_) { |
| 264 | // Acquire the lock to prevent races between readers of 'is_cancelled_' and this |
| 265 | // writing thread (e.g. IMPALA-9611) - the caller should read 'is_cancelled_' while |
| 266 | // holding the lock. Drop it before signalling the CV so that a blocked thread can |
| 267 | // immediately acquire the mutex when it wakes up. |
| 268 | { |
| 269 | lock_guard<mutex> l(*entry.first); |
| 270 | } |
| 271 | entry.second->NotifyAll(); |
| 272 | } |
| 273 | for (CyclicBarrier* cb : cancellation_cbs_) { |
| 274 | cb->Cancel(Status::CancelledInternal("RuntimeState::Cancel()")); |
| 275 | } |
| 276 | } |
| 277 | |
| 278 | } |
| 279 | |
| 280 | void RuntimeState::AddCancellationCV(mutex* mutex, ConditionVariable* cv) { |
| 281 | lock_guard<SpinLock> l(cancellation_cvs_lock_); |