Status TransferEngine::submitTransfer(
    BatchID batch_id, const std::vector<TransferRequest>& entries) {
    return impl_->submitTransfer(batch_id, entries);
}

Status RdmaTransport::submitTransfer(
    BatchID batch_id, const std::vector<TransferRequest>& entries) {
    auto& batch_desc = *((BatchDesc*)(batch_id));

    // 容量检查
    if (batch_desc.task_list.size() + entries.size() > batch_desc.batch_size) {
        return Status::InvalidArgument("Exceed capacity");
    }

    size_t task_id = batch_desc.task_list.size();
    batch_desc.task_list.resize(task_id + entries.size());
    std::vector<TransferTask*> task_list;
    for (auto& task : batch_desc.task_list)
        task_list.push_back(&task);
    return submitTransferTask(task_list);
}

Status RdmaTransport::submitTransferTask(
    const std::vector<TransferTask*>& task_list) {
    const size_t kBlockSize = globalConfig().slice_size;    // 默认 64KB
    const int kMaxRetryCount = globalConfig().retry_cnt;
    const size_t kFragmentSize = globalConfig().fragment_limit;

    for (size_t index = 0; index < task_list.size(); ++index) {
        auto& task = *task_list[index];
        auto& request = *task.request;

        // 为本地源地址选择设备
        auto request_buffer_id = -1, request_device_id = -1;
        selectDevice(local_segment_desc.get(), (uint64_t)request.source,
                     request.length, request_buffer_id, request_device_id);

        // 将大请求分割为固定大小的 Slice
        for (uint64_t offset = 0; offset < request.length; offset += kBlockSize) {
            Slice* slice = getSliceCache().allocate();

            // 最后一个 slice 可能合并尾部碎片
            bool merge_final_slice =
                request.length - offset <= kBlockSize + kFragmentSize;

            slice->source_addr = (char*)request.source + offset;
            slice->length = merge_final_slice ?
                request.length - offset : kBlockSize;
            slice->opcode = request.opcode;
            slice->rdma.dest_addr = request.target_offset + offset;
            slice->rdma.retry_cnt = request.advise_retry_cnt;
            slice->rdma.max_retry_cnt = kMaxRetryCount;
            slice->task = &task;
            slice->target_id = request.target_id;
            slice->status = Slice::PENDING;

            task.slice_list.push_back(slice);

            // 为目标地址选择 RDMA 设备
            // ...selectDevice for target...
            // 按 context 分组: slices_to_post[context].push_back(slice)
        }
    }

    // 批量提交到各 RdmaContext
    for (auto& [context, slices] : slices_to_post) {
        context->submitPostSend(slices);
    }
}

int TransferEngine::registerLocalMemory(void* addr, size_t length,
                                        const std::string& location,
                                        bool remote_accessible,
                                        bool update_metadata) {
    return impl_->registerLocalMemory(addr, length, location, remote_accessible,
                                      update_metadata);
}

int RdmaTransport::registerLocalMemoryInternal(void* addr, size_t length,
                                               const std::string& name,
                                               bool remote_accessible,
                                               bool update_metadata,
                                               bool force_sequential) {
    BufferDesc buffer_desc;
    const int kBaseAccessRights = IBV_ACCESS_LOCAL_WRITE |
                                  IBV_ACCESS_REMOTE_WRITE |
                                  IBV_ACCESS_REMOTE_READ;
    static int access_rights = kBaseAccessRights;
    if (MCIbRelaxedOrderingEnabled) {
        access_rights |= IBV_ACCESS_RELAXED_ORDERING;
    }

MCIbRelaxedOrderingEnabled = has_ibv_reg_mr_iova2();

// 当内存 >= 4GB 且 CPU 核心数 >= 4 时执行并行预触摸
bool do_pre_touch = context_list_.size() > 0 &&
                    std::thread::hardware_concurrency() >= 4 &&
                    length >= (size_t)4 * 1024 * 1024 * 1024;
if (do_pre_touch) {
    int ret = preTouchMemory(addr, length);
    // ...
}

// 并行注册判断
int use_parallel_reg = 0;
if (!force_sequential) {
    use_parallel_reg = globalConfig().parallel_reg_mr;
    if (use_parallel_reg == -1) {
        // 自动模式: 多 context 且已 pre-touch 时并行
        use_parallel_reg = context_list_.size() > 1 && do_pre_touch;
    }
}

if (use_parallel_reg) {
    // 为每个 RdmaContext 启动独立线程
    std::vector<std::thread> reg_threads;
    for (size_t i = 0; i < context_list_.size(); ++i) {
        reg_threads.emplace_back([this, &ret_codes, i, addr, length, ar]() {
            ret_codes[i] = context_list_[i]->registerMemoryRegion(addr, length, ar);
        });
    }
    // 等待所有线程完成
} else {
    // 串行注册
    for (size_t i = 0; i < context_list_.size(); ++i) {
        int ret = context_list_[i]->registerMemoryRegion(addr, length, access_rights);
    }
}

// 收集所有 context 的 lkey/rkey
for (auto& context : context_list_) {
    buffer_desc.lkey.push_back(context->lkey(addr));
    buffer_desc.rkey.push_back(context->rkey(addr));
}

// 自动检测内存位置 (CPU NUMA 节点或 GPU 设备)
if (name == kWildcardLocation) {
    bool only_first_page = true;
    const std::vector<MemoryLocationEntry> entries =
        getMemoryLocation(addr, length, only_first_page);
    buffer_desc.name = entries[0].location;  // 如 "cpu:0" 或 "cuda:1"
} else {
    buffer_desc.name = name;
}

buffer_desc.addr = (uint64_t)addr;
buffer_desc.length = length;
// 注册到元数据服务
int rc = metadata_->addLocalMemoryBuffer(buffer_desc, update_metadata);

int Topology::discover(const std::vector<std::string>& filter) {
    matrix_.clear();

    // 1. 列举所有 InfiniBand 设备
    auto all_hca = listInfiniBandDevices(filter);

    // 2. CPU 拓扑发现
    for (auto& ent : discoverCpuTopology(all_hca)) {
        matrix_[ent.name] = ent;
    }

    // 3. GPU 拓扑发现 (如果编译了 CUDA/MUSA/HIP)
#if defined(USE_CUDA) || defined(USE_MUSA) || defined(USE_HIP)
    for (auto& ent : discoverCudaTopology(all_hca)) {
        matrix_[ent.name] = ent;
    }
#endif

    // 4. 解析为索引化的 resolved_matrix_
    return resolve();
}

static std::vector<InfinibandDevice> listInfiniBandDevices(
    const std::vector<std::string>& filter) {
    int num_devices = 0;
    struct ibv_device** device_list = ibv_get_device_list(&num_devices);
    // ...

    for (int i = 0; i < num_devices; ++i) {
        std::string device_name = ibv_get_device_name(device_list[i]);

        // 如果提供了 filter,只保留指定设备
        if (!filter.empty() && std::find(...) == filter.end())
            continue;

        // 通过 sysfs 解析 PCI 总线地址
        // /sys/class/infiniband/mlx5_X -> /sys/devices/pciXXXX:XX/...
        snprintf(path, sizeof(path), "/sys/class/infiniband/%s/../..",
                 device_name.c_str());
        realpath(path, resolved_path);
        std::string pci_bus_id = basename(resolved_path);

        // 读取 NUMA 节点号
        snprintf(path, sizeof(path), "%s/numa_node", resolved_path);
        std::ifstream(path) >> numa_node;

        devices.push_back(InfinibandDevice{
            .name = device_name,
            .pci_bus_id = pci_bus_id,
            .numa_node = numa_node
        });
    }
}

static std::vector<TopologyEntry> discoverCpuTopology(
    const std::vector<InfinibandDevice>& all_hca) {
    DIR* dir = opendir("/sys/devices/system/node");
    // 遍历 /sys/devices/system/node/nodeN

    while ((entry = readdir(dir))) {
        int node_id = atoi(entry->d_name + strlen("node"));

        // HCA 与 CPU NUMA 节点同节点 -> preferred
        // HCA 与 CPU NUMA 节点不同节点 -> available
        for (const auto& hca : all_hca) {
            if (hca.numa_node == node_id) {
                preferred_hca.push_back(hca.name);
            } else {
                avail_hca.push_back(hca.name);
            }
        }

        topology.push_back(TopologyEntry{
            .name = "cpu:" + std::to_string(node_id),
            .preferred_hca = preferred_hca,
            .avail_hca = avail_hca
        });
    }
}

static std::vector<TopologyEntry> discoverCudaTopology(
    const std::vector<InfinibandDevice>& all_hca) {
    int device_count;
    cudaGetDeviceCount(&device_count);

    for (int i = 0; i < device_count; i++) {
        char pci_bus_id[20];
        cudaDeviceGetPCIBusId(pci_bus_id, sizeof(pci_bus_id), i);

        // 1. 首先筛选同 NUMA 节点的 HCA
        for (const auto& hca : all_hca) {
            if (isSameNumaNode(hca.pci_bus_id.c_str(), pci_bus_id)) {
                sameNuma_hca.push_back(hca);
            }
        }

        // 2. 在同 NUMA 候选中,按 PCI 拓扑距离排序
        for (const auto& hca : candidate_preferred_hca) {
            int distance = getPciDistance(hca.pci_bus_id.c_str(), pci_bus_id);
            // 距离最小的 HCA 为 preferred
        }

        topology.push_back(TopologyEntry{
            .name = "cuda:" + std::to_string(i),
            .preferred_hca = preferred_hca,
            .avail_hca = avail_hca
        });
    }
}

int Topology::resolve() {
    resolved_matrix_.clear();
    hca_list_.clear();
    std::map<std::string, int> hca_id_map;
    int next_hca_map_index = 0;

    for (auto& entry : matrix_) {
        for (auto& hca : entry.second.preferred_hca) {
            if (!hca_id_map.count(hca)) {
                hca_list_.push_back(hca);
                hca_id_map[hca] = next_hca_map_index++;

                // 通配符位置 "*" 包含所有 HCA 作为 preferred
                resolved_matrix_[kWildcardLocation].preferred_hca.push_back(
                    hca_id_map[hca]);
            }
            resolved_matrix_[entry.first].preferred_hca.push_back(hca_id_map[hca]);
        }
        // avail_hca 同理处理...
    }
    return 0;
}

int Topology::selectDevice(const std::string storage_type, int retry_count) {
    auto& entry = resolved_matrix_[storage_type];

    if (retry_count == 0) {
        int rand_value;
        if (use_round_robin_) {
            thread_local int tl_counter = 0;
            rand_value = tl_counter;
            tl_counter = (tl_counter + 1) % 10000;
        } else {
            rand_value = SimpleRandom::Get().next();
        }

        // 优先从 preferred_hca 中选择
        if (!entry.preferred_hca.empty())
            return entry.preferred_hca[rand_value % entry.preferred_hca.size()];
        else
            return entry.avail_hca[rand_value % entry.avail_hca.size()];
    } else {
        // 重试时按序选择不同设备
        size_t index = (retry_count - 1) %
                       (entry.preferred_hca.size() + entry.avail_hca.size());
        // ...
    }
}