Add NIXL backend - #6016

void NixlBackend::Impl::waitTransfer(NixlTransferHandle& handle) {
 NVF_ERROR(handle.isValid(), "Cannot wait on an invalid handle");
 NVF_ERROR(handle.impl_->posted, "Transfer has not been posted yet");
 // TODO - check this spin loop
 NixlXferStatus xfer_status;
 do {
 xfer_status = getTransferStatus(handle);
 NVF_ERROR(
 xfer_status != NixlXferStatus::kError,
 "NIXL transfer completed with an error");
 } while (xfer_status == NixlXferStatus::kInProgress);
 handle.impl_->posted = false;
}

void NixlBackend::Impl::exchangeMetadata() {
 nixl_blob_t local_md;
 nixl_status_t md_status = agent_->getLocalMD(local_md);
 NVF_ERROR(
 md_status == NIXL_SUCCESS,
 "NIXL getLocalMD failed with status ",
 static_cast<int>(md_status));
 auto* store = communicator_.getTcpStore();
 const auto my_rank = communicator_.deviceId();
 const auto world_size = communicator_.size();
 std::string md_key_prefix = "nixl_agent_md_rank_";
 store->set(
 md_key_prefix + std::to_string(my_rank),
 std::vector<uint8_t>(local_md.begin(), local_md.end()));
 for (int64_t rank = 0; rank < world_size; ++rank) {
 if (rank == my_rank) {
 continue;
 }
 // Fetch & load MD
 auto bytes = store->get(md_key_prefix + std::to_string(rank));
 nixl_blob_t remote_md(bytes.begin(), bytes.end());
 std::string remote_agent_name;
 nixl_status_t status = agent_->loadRemoteMD(remote_md, remote_agent_name);
 NVF_ERROR(
 status == NIXL_SUCCESS,
 "NIXL loadRemoteMD failed for rank ",
 rank,
 " with status ",
 static_cast<int>(status));
 }
 // Barrier before deleting keys so no rank reads a deleted key.
 communicator_.barrier();
 store->deleteKey(md_key_prefix + std::to_string(my_rank));
 metadata_exchanged_ = true;
}

// Probe: verify that VRAM (CUDA GPU memory) is actually usable with
// the UCX backend. Some UCX installations lack CUDA support, causing
// registerMem to silently misclassify VRAM as host memory. We detect
// this by registering a small buffer and asking NIXL to prepare a
// local descriptor list for VRAM -- if no backend claims VRAM, the
// probe fails and we mark the backend as unavailable.
{
 constexpr int64_t kProbeBytes = 1;
 auto probe = at::empty(
 {kProbeBytes},
 at::TensorOptions().dtype(at::kByte).device(
 at::kCUDA, communicator.deviceId()));
 size_t nbytes = static_cast<size_t>(probe.nbytes());
 uintptr_t addr = reinterpret_cast<uintptr_t>(probe.data_ptr());
 uint32_t dev_idx = static_cast<uint32_t>(probe.device().index());
 NVF_ERROR(nbytes > 0, "NIXL probe: unexpected zero-byte tensor");
 NVF_ERROR(addr != 0, "NIXL probe: null data pointer");
 nixl_reg_dlist_t reg_dlist(VRAM_SEG);
 reg_dlist.addDesc({addr, nbytes, static_cast<uint64_t>(dev_idx)});
 nixl_status_t reg_status = impl->agent_->registerMem(reg_dlist);
 if (reg_status != NIXL_SUCCESS) {
 return nullptr;
 }
 nixl_xfer_dlist_t xfer_dlist(VRAM_SEG);
 xfer_dlist.addDesc({addr, nbytes, static_cast<uint64_t>(dev_idx)});
 nixlDlistH* dlist_handle = nullptr;
 nixl_status_t prep_status =
 impl->agent_->prepXferDlist(NIXL_INIT_AGENT, xfer_dlist, dlist_handle);
 if (dlist_handle) {
 impl->agent_->releasedDlistH(dlist_handle);
 }
 impl->agent_->deregisterMem(reg_dlist);
 if (prep_status != NIXL_SUCCESS) {
 return nullptr;
 }
}