Method RecvAsync

tensorflow/compiler/jit/kernels/async_io_rendezvous.cc:59–95

}

void AsyncIoRendezvous::RecvAsync(uint64 key_hash, DoneCallback done) {
  VLOG(2) << "Recv " << this << " " << key_hash;

  MutexedItemQueue* mq = nullptr;
  {
    mutex_lock l(mu_);
    mq = table_.at(key_hash);
  }
  CHECK(mq);
  ItemQueue* queue = &mq->queue;
  mutex& mu = mq->mu;
  mu.lock();
  if (queue->empty() || !queue->front()->IsSendValue()) {
    // There is no message to pick up.
    // Only recv-related fields need to be filled.
    VLOG(2) << "Enqueue Recv Item (key hash:" << key_hash << "). ";
    Item* item = new Item;
    item->waiter = std::move(done);

    queue->push_back(item);
    mu.unlock();
    return;
  }

  VLOG(2) << "Consume Send Item (key hash:" << key_hash << "). ";
  // A message has already arrived and is queued in the table under
  // this key.  Consumes the message and invokes the done closure.
  Item* item = queue->front();
  queue->pop_front();
  mu.unlock();

  // Invokes the done() by invoking its done closure, outside scope
  // of the table lock.
  CHECK(item->IsSendValue());
  done(Status::OK(), item->value);
  delete item;
}

/*static*/ AsyncIoRendezvous* GetXlaAsyncIORendezvous() {
  static AsyncIoRendezvous* self = new AsyncIoRendezvous();