| 1257 | |
| 1258 | template <class PropagatorStateType> |
| 1259 | bool ExecutorState<PropagatorStateType>::NodeDone( |
| 1260 | const Status& s, TaggedNodeSeq* ready, NodeExecStatsInterface* stats, |
| 1261 | TaggedNodeReadyQueue* inline_ready) { |
| 1262 | if (stats) { |
| 1263 | nodestats::SetAllEnd(stats); |
| 1264 | tracing::CallingContext::Pop(); |
| 1265 | DCHECK_NE(stats_collector_, nullptr); |
| 1266 | stats->Done(immutable_state_.params().device->name()); |
| 1267 | } |
| 1268 | |
| 1269 | if (TF_PREDICT_TRUE(s.ok())) { |
| 1270 | const size_t ready_size = ready->size(); |
| 1271 | if (ready_size == 0) { |
| 1272 | return num_outstanding_ops_.fetch_sub(1) == 1; |
| 1273 | } else { |
| 1274 | // NOTE: Avoid touching the atomic counter if only one node becomes ready. |
| 1275 | if (ready_size > 1) { |
| 1276 | num_outstanding_ops_.fetch_add(ready_size - 1, |
| 1277 | std::memory_order_relaxed); |
| 1278 | } |
| 1279 | |
| 1280 | // Schedule the ready nodes in 'ready'. |
| 1281 | ScheduleReady(ready, inline_ready); |
| 1282 | |
| 1283 | return false; |
| 1284 | } |
| 1285 | } else { |
| 1286 | bool abort_run = false; |
| 1287 | |
| 1288 | // Some error happened. This thread of computation is done. |
| 1289 | { |
| 1290 | mutex_lock l(mu_); |
| 1291 | if (status_.ok()) { |
| 1292 | // If this is the first node to fail in this run, we are responsible for |
| 1293 | // aborting all other execution in the step. |
| 1294 | abort_run = true; |
| 1295 | |
| 1296 | // If execution has been cancelled, mark cancelled or aborted errors as |
| 1297 | // being derived. Note that the original node that fails might also |
| 1298 | // trigger cancellation, and here we make sure the original error is |
| 1299 | // exposed to users and not buried as a derived error. |
| 1300 | if (cancellation_manager_ && cancellation_manager_->IsCancelled() && |
| 1301 | (errors::IsCancelled(s) || errors::IsAborted(s))) { |
| 1302 | status_ = StatusGroup::MakeDerived(s); |
| 1303 | } else { |
| 1304 | status_ = s; |
| 1305 | } |
| 1306 | } |
| 1307 | } |
| 1308 | |
| 1309 | if (abort_run) { |
| 1310 | TRACEPRINTF("StartAbort: %s", s.ToString().c_str()); |
| 1311 | if (cancellation_manager_) { |
| 1312 | // Only log when the abort happens during the actual run time. |
| 1313 | // Use VLOG instead of LOG(warning) because error status is expected |
| 1314 | // when the executor is run under the grappler optimization phase or |
| 1315 | // when iterating through a tf.data input pipeline. |
| 1316 | VLOG(1) << "[" << immutable_state_.params().device->name() |
nothing calls this directly
no test coverage detected