| 841 | } |
| 842 | |
| 843 | void BaseRemoteRendezvous::StartAbort(const Status& s) { |
| 844 | CHECK(!s.ok()); |
| 845 | // Use a "derived" status as the status for the rendezvous. Derived |
| 846 | // status messages are ignored when aggregating errors across devices: this |
| 847 | // allows us to prefer our original status message over any cancellation |
| 848 | // related errors. |
| 849 | Status derived_status = StatusGroup::MakeDerived(s); |
| 850 | |
| 851 | local_->StartAbort(derived_status); |
| 852 | { |
| 853 | // Aborts all active RecvTensor calls. |
| 854 | mutex_lock l(mu_); |
| 855 | if (status_.ok()) { |
| 856 | status_ = derived_status; |
| 857 | for (auto& entry : active_) { |
| 858 | entry.first->StartAbort(derived_status); |
| 859 | entry.second(); |
| 860 | } |
| 861 | active_.clear(); |
| 862 | } |
| 863 | flow_control_num_ = 0; |
| 864 | flow_control_counters_.clear(); |
| 865 | } |
| 866 | flow_control_cv_.notify_all(); |
| 867 | } |
| 868 | |
| 869 | void BaseRemoteRendezvous::RegisterCall(BaseRecvTensorCall* call, |
| 870 | const Rendezvous::Args& args) { |