| 269 | } |
| 270 | |
| 271 | void RingAlg::StartAbort(const Status& s) { |
| 272 | // In abort mode we stop issuing additional ProvideBuf |
| 273 | // and ConsumeBuf calls, but we need to wait for all of the |
| 274 | // outstanding callbacks to be invoked before quitting. |
| 275 | bool abort_started = false; |
| 276 | { |
| 277 | mutex_lock l(status_mu_); |
| 278 | if (status_.ok()) { |
| 279 | LOG(ERROR) << "Aborting Ring" << name_ << " with " << s; |
| 280 | abort_started = true; |
| 281 | status_.Update(s); |
| 282 | } |
| 283 | } |
| 284 | // If this is the initial entry to abort mode then invoke StartAbort |
| 285 | // on the CollectiveExecutor that invoked us. That should start |
| 286 | // cancellation on all of the outstanding CollectiveRemoteAccess |
| 287 | // actions. |
| 288 | if (abort_started) { |
| 289 | col_ctx_->col_exec->StartAbort(s); |
| 290 | } |
| 291 | } |
| 292 | |
| 293 | void RingAlg::Finish(bool ok) { |
| 294 | if (ok) { |