Method Send

tensorflow/core/framework/rendezvous.cc:194–244

  explicit LocalRendezvousImpl() {}

  Status Send(const ParsedKey& key, const Args& send_args, const Tensor& val,
              const bool is_dead) override {
    uint64 key_hash = KeyHash(key.FullKey());
    VLOG(2) << "Send " << this << " " << key_hash << " " << key.FullKey();

    mu_.lock();
    if (!status_.ok()) {
      // Rendezvous has been aborted.
      Status s = status_;
      mu_.unlock();
      return s;
    }

    ItemQueue* queue = &table_[key_hash];
    if (queue->empty() || queue->front()->IsSendValue()) {
      // There is no waiter for this message. Append the message
      // into the queue. The waiter will pick it up when arrives.
      // Only send-related fields need to be filled.
      VLOG(2) << "Enqueue Send Item (key:" << key.FullKey() << "). ";
      Item* item = new Item;
      item->value = val;
      item->is_dead = is_dead;
      item->send_args = send_args;
      if (item->send_args.device_context) {
        item->send_args.device_context->Ref();
      }
      queue->push_back(item);
      mu_.unlock();
      return Status::OK();
    }

    VLOG(2) << "Consume Recv Item (key:" << key.FullKey() << "). ";
    // There is an earliest waiter to consume this message.
    Item* item = queue->front();

    // Delete the queue when the last element has been consumed.
    if (queue->size() == 1) {
      VLOG(2) << "Clean up Send/Recv queue (key:" << key.FullKey() << "). ";
      table_.erase(key_hash);
    } else {
      queue->pop_front();
    }
    mu_.unlock();

    // Notify the waiter by invoking its done closure, outside the
    // lock.
    DCHECK(!item->IsSendValue());
    item->waiter(Status::OK(), send_args, item->recv_args, val, is_dead);
    delete item;
    return Status::OK();
  }

  // support send ref tensor to local rendezvous
  Status Send(const ParsedKey& key, const Args& send_args,