Method TryAttemptLocked

tensorflow/core/kernels/queue_base.cc:264–312

}

bool QueueBase::TryAttemptLocked(Action action,
                                 std::vector<CleanUp>* clean_up) {
  std::deque<Attempt>* attempts =
      action == kEnqueue ? &enqueue_attempts_ : &dequeue_attempts_;

  bool progress = false;
  bool done = false;
  while (!done && !attempts->empty()) {
    if (attempts->front().is_cancelled) {
      if (action == kEnqueue) {
        if (closed_) {
          VLOG(1) << "Skipping cancelled enqueue attempt";
        } else {
          LOG(WARNING)
              << name_
              << ": Skipping cancelled enqueue attempt with queue not closed";
        }
      } else {
        if (closed_) {
          VLOG(1) << "Skipping cancelled dequeue attempt";
        } else {
          LOG(WARNING)
              << name_
              << ": Skipping cancelled dequeue attempt with queue not closed";
        }
      }
      attempts->pop_front();
    } else {
      Attempt* cur_attempt = &attempts->front();
      switch (cur_attempt->run_callback(cur_attempt)) {
        case kNoProgress:
          done = true;
          break;
        case kProgress:
          done = true;
          progress = true;
          break;
        case kComplete:
          progress = true;
          clean_up->emplace_back(std::move(cur_attempt->done_callback),
                                 cur_attempt->cancellation_token,
                                 cur_attempt->context->cancellation_manager());
          attempts->pop_front();
          break;
      }
    }
  }
  return progress;
}

void QueueBase::FlushUnlocked() {
  std::vector<CleanUp> clean_up;