Method AcquireAsync

tensorflow/core/kernels/mutex_ops.cc:121–168

  };

  void AcquireAsync(
      OpKernelContext* c,
      std::function<void(const Status& s, SharedLockReleaser lock)> fn) {
    CancellationManager* cm = c->cancellation_manager();
    CancellationToken token{};
    bool* cancelled = nullptr;
    if (cm) {
      cancelled = new bool(false);  // GUARDED_BY(mu_);
      token = cm->get_cancellation_token();
      const bool already_cancelled =
          !cm->RegisterCallback(token, [this, cancelled]() {
            mutex_lock lock(mu_);
            *cancelled = true;
            cv_.notify_all();
          });
      if (already_cancelled) {
        delete cancelled;
        fn(errors::Cancelled("Lock acquisition cancelled."),
           SharedLockReleaser{nullptr});
        return;
      }
    }
    thread_pool_->Schedule(std::bind(
        [this, cm, cancelled,
         token](std::function<void(const Status& s, SharedLockReleaser&& lock)>
                    fn_) {
          bool local_locked;
          {
            mutex_lock lock(mu_);
            while (locked_ && !(cancelled && *cancelled)) {
              cv_.wait(lock);
            }
            local_locked = locked_ = !(cancelled && *cancelled);
          }
          if (cm) {
            cm->DeregisterCallback(token);
            delete cancelled;
          }
          if (local_locked) {  // Not cancelled.
            fn_(Status::OK(),
                SharedLockReleaser{std::make_shared<LockReleaser>(this)});
          } else {
            fn_(errors::Cancelled("Lock acquisition cancelled."),
                SharedLockReleaser{nullptr});
          }
        },
        std::move(fn)));
  }

 private:
  mutex mu_;