Function EnqueueTensorCollective

tensorflow/contrib/mpi_collectives/mpi_ops.cc:839–852

This function (called from the callback set up in MPIAll*Op::ComputeAsync) only adds the op's record into the local op queue (to track the op's progress), and sends a message to the coordinator indicating that this rank is ready to begin. The MPI background thread will handle the MPI message.

// progress), and sends a message to the coordinator indicating that this rank
// is ready to begin. The MPI background thread will handle the MPI message.
void EnqueueTensorCollective(CollectiveOpRecord record,
                             MPIRequest::RequestType rtype) {
  const Tensor* input_tensor = record.in_t;
  MPIRequest message;
  message.set_request_rank(record.rank);
  message.set_tensor_name(record.name);
  message.set_tensor_type(record.dtype);
  message.set_request_type(rtype);
  input_tensor->shape().AsProto(message.mutable_tensor_shape());

  mutex_lock guard(mpi_global.mu);
  mpi_global.tensor_table.emplace(record.name, record);
  mpi_global.message_queue.push(message);
}

}  // namespace