[WIP] [Feat] Ffts D2D transport dev

ucm/transport/p2p/CMakeLists.txt

@@ -0,0 +1,3 @@
+if(RUNTIME_ENVIRONMENT STREQUAL "ascend")
+    add_subdirectory(ffts)
+endif()

ucm/transport/p2p/ffts/CMakeLists.txt

@@ -0,0 +1,120 @@
+# Copyright (c) 2026 Huawei Technologies Co., Ltd.
+
+option(BUILD_FFTS_BENCHMARKS "build FFTS transport benchmark executables." OFF)
+
+if(DEFINED ENV{ASCEND_HOME_PATH})
+    set(Ascend_ROOT $ENV{ASCEND_HOME_PATH})
+elseif(DEFINED ENV{ASCEND_CUSTOM_PATH})
+    set(Ascend_ROOT $ENV{ASCEND_CUSTOM_PATH}/latest)
+else()
+    set(Ascend_ROOT /usr/local/Ascend/ascend-toolkit/latest)
+endif()
+
+find_path(FFTS_ASCEND_INCLUDE_DIR
+    NAMES acl/acl.h
+    HINTS
+        ${Ascend_ROOT}/include
+        ${Ascend_ROOT}/aarch64-linux/include
+        ${Ascend_ROOT}/arm64-linux/include
+    NO_DEFAULT_PATH
+)
+
+find_path(FFTS_RUNTIME_INCLUDE_DIR
+    NAMES runtime/rt_ffts_plus.h
+    HINTS
+        ${Ascend_ROOT}/pkg_inc/runtime
+        ${Ascend_ROOT}/aarch64-linux/pkg_inc/runtime
+        ${Ascend_ROOT}/arm64-linux/pkg_inc/runtime
+    NO_DEFAULT_PATH
+)
+
+find_path(FFTS_TOOLCHAIN_INCLUDE_DIR
+    NAMES toolchain/prof_api.h
+    HINTS
+        ${Ascend_ROOT}/pkg_inc
+        ${Ascend_ROOT}/aarch64-linux/pkg_inc
+        ${Ascend_ROOT}/arm64-linux/pkg_inc
+        ${Ascend_ROOT}/include/experiment/msprof
+        ${Ascend_ROOT}/aarch64-linux/include/experiment/msprof
+        ${Ascend_ROOT}/arm64-linux/include/experiment/msprof
+    NO_DEFAULT_PATH
+)
+
+find_path(FFTS_PROFILING_INCLUDE_DIR
+    NAMES prof_common.h
+    HINTS
+        ${Ascend_ROOT}/pkg_inc/profiling
+        ${Ascend_ROOT}/aarch64-linux/pkg_inc/profiling
+        ${Ascend_ROOT}/arm64-linux/pkg_inc/profiling
+        ${Ascend_ROOT}/include/experiment/msprof/toolchain
+        ${Ascend_ROOT}/aarch64-linux/include/experiment/msprof/toolchain
+        ${Ascend_ROOT}/arm64-linux/include/experiment/msprof/toolchain
+    NO_DEFAULT_PATH
+)
+
+find_library(FFTS_ASCENDCL_LIBRARY
+    NAMES ascendcl
+    HINTS
+        ${Ascend_ROOT}/lib64
+        ${Ascend_ROOT}/aarch64-linux/lib64
+        ${Ascend_ROOT}/arm64-linux/lib64
+    NO_DEFAULT_PATH
+)
+
+find_library(FFTS_RUNTIME_LIBRARY
+    NAMES runtime
+    HINTS
+        ${Ascend_ROOT}/lib64
+        ${Ascend_ROOT}/runtime/lib64
+        ${Ascend_ROOT}/aarch64-linux/lib64
+        ${Ascend_ROOT}/arm64-linux/lib64
+    NO_DEFAULT_PATH
+)
+
+if(NOT FFTS_ASCEND_INCLUDE_DIR OR NOT FFTS_RUNTIME_INCLUDE_DIR OR NOT FFTS_TOOLCHAIN_INCLUDE_DIR OR
+   NOT FFTS_PROFILING_INCLUDE_DIR OR NOT FFTS_ASCENDCL_LIBRARY OR NOT FFTS_RUNTIME_LIBRARY)
+    message(FATAL_ERROR "Ascend CANN headers/libraries required for ucm_transport_ffts were not found")
+endif()
+
+find_package(Threads REQUIRED)
+
+add_library(ucm_transport_ffts STATIC
+    src/ffts_transport.cpp
+    src/ffts_engine.cpp
+)
+
+target_compile_features(ucm_transport_ffts PUBLIC cxx_std_17)
+
+target_include_directories(ucm_transport_ffts
+    PUBLIC
+        ${CMAKE_CURRENT_SOURCE_DIR}/include
+    PRIVATE
+        ${CMAKE_CURRENT_SOURCE_DIR}
+)
+
+target_include_directories(ucm_transport_ffts SYSTEM
+    PRIVATE
+        ${FFTS_ASCEND_INCLUDE_DIR}
+        ${FFTS_RUNTIME_INCLUDE_DIR}
+        ${FFTS_TOOLCHAIN_INCLUDE_DIR}
+        ${FFTS_PROFILING_INCLUDE_DIR}
+)
+
+target_link_libraries(ucm_transport_ffts
+    PUBLIC
+        infra_status
+    PRIVATE
+        Threads::Threads
+        ${CMAKE_DL_LIBS}
+        m
+        ${FFTS_ASCENDCL_LIBRARY}
+        ${FFTS_RUNTIME_LIBRARY}
+)
+
+if(BUILD_UNIT_TESTS)
+    add_subdirectory(test)
+endif()
+
+if(BUILD_FFTS_BENCHMARKS)
+    add_subdirectory(bench)
+endif()

ucm/transport/p2p/ffts/bench/CMakeLists.txt

@@ -0,0 +1,16 @@
+# Copyright (c) 2026 Huawei Technologies Co., Ltd.
+
+add_executable(ucm_transport_ffts_bench
+    ffts_bench_options.cpp
+    ffts_bench_runtime.cpp
+    ffts_bench_runner.cpp
+    ffts_transport_bench.cpp
+)
+
+target_link_libraries(ucm_transport_ffts_bench PRIVATE
+    ucm_transport_ffts
+)
+
+target_include_directories(ucm_transport_ffts_bench SYSTEM PRIVATE
+    ${FFTS_ASCEND_INCLUDE_DIR}
+)

ucm/transport/p2p/ffts/bench/ffts_bench_options.cpp

@@ -0,0 +1,144 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2026 Huawei Technologies Co., Ltd. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ * */
+#include "ffts_bench_options.h"
+
+#include <cctype>
+#include <cstdlib>
+#include <iostream>
+#include <limits>
+#include <stdexcept>
+#include <string>
+
+namespace UC::Transport::Ffts::Bench {
+namespace {
+size_t ParseBytes(const std::string& value)
+{
+    if (value.empty()) { throw std::invalid_argument("empty size value"); }
+
+    std::string number = value;
+    size_t multiplier = 1;
+    const auto suffix = static_cast<char>(std::tolower(static_cast<unsigned char>(value.back())));
+    if (suffix == 'k' || suffix == 'm' || suffix == 'g') {
+        number = value.substr(0, value.size() - 1);
+        multiplier = suffix == 'k' ? kKiB : (suffix == 'm' ? kMiB : kGiB);
+    }
+
+    size_t consumed = 0;
+    const auto parsed = std::stoull(number, &consumed, 0);
+    if (consumed != number.size()) { throw std::invalid_argument("invalid size value: " + value); }
+    if (parsed > std::numeric_limits<size_t>::max() / multiplier) {
+        throw std::overflow_error("size value overflows: " + value);
+    }
+    return static_cast<size_t>(parsed) * multiplier;
+}
+
+size_t ParseCount(const std::string& value)
+{
+    size_t consumed = 0;
+    const auto parsed = std::stoull(value, &consumed, 0);
+    if (consumed != value.size()) { throw std::invalid_argument("invalid count value: " + value); }
+    return static_cast<size_t>(parsed);
+}
+
+std::vector<size_t> ParseList(const std::string& value, bool parseBytes)
+{
+    std::vector<size_t> result;
+    size_t begin = 0;
+    while (begin <= value.size()) {
+        const auto end = value.find(',', begin);
+        const auto token = value.substr(begin, end == std::string::npos ? std::string::npos : end - begin);
+        if (token.empty()) { throw std::invalid_argument("empty list item: " + value); }
+        result.push_back(parseBytes ? ParseBytes(token) : ParseCount(token));
+        if (end == std::string::npos) { break; }
+        begin = end + 1;
+    }
+    return result;
+}
+
+Scenario ParseScenario(const std::string& value)
+{
+    if (value == "all") { return Scenario::All; }
+    if (value == "single") { return Scenario::Single; }
+    if (value == "batch") { return Scenario::Batch; }
+    throw std::invalid_argument("invalid scenario: " + value);
+}
+
+void PrintUsage(const char* program)
+{
+    std::cout << "Usage: " << program << " [options]\n"
+              << "Options:\n"
+              << "  --device N              Ascend device id, default 0\n"
+              << "  --warmup N              Warmup iterations per case, default 10\n"
+              << "  --iters N               Timed iterations per case, default 100\n"
+              << "  --scenario NAME         all, single, or batch; default all\n"
+              << "  --min-bytes SIZE        Minimum single-copy size, default 4K\n"
+              << "  --max-bytes SIZE        Maximum single-copy size, default 256M\n"
+              << "  --batch-counts LIST     Comma-separated batch counts, default 4,16,64,128\n"
+              << "  --batch-chunks LIST     Comma-separated chunk sizes, default 4K,16K,64K,256K,1M\n"
+              << "  --help                  Show this help text\n";
+}
+}  // namespace
+
+Options ParseOptions(int argc, char** argv)
+{
+    Options options;
+    for (int i = 1; i < argc; ++i) {
+        const std::string arg = argv[i];
+        auto requireValue = [&](const char* name) {
+            if (i + 1 >= argc) { throw std::invalid_argument(std::string(name) + " requires a value"); }
+            return std::string(argv[++i]);
+        };
+
+        if (arg == "--help") {
+            PrintUsage(argv[0]);
+            std::exit(0);
+        } else if (arg == "--device") {
+            options.deviceId = static_cast<int32_t>(ParseCount(requireValue("--device")));
+        } else if (arg == "--warmup") {
+            options.warmup = ParseCount(requireValue("--warmup"));
+        } else if (arg == "--iters") {
+            options.iterations = ParseCount(requireValue("--iters"));
+        } else if (arg == "--scenario") {
+            options.scenario = ParseScenario(requireValue("--scenario"));
+        } else if (arg == "--min-bytes") {
+            options.minBytes = ParseBytes(requireValue("--min-bytes"));
+        } else if (arg == "--max-bytes") {
+            options.maxBytes = ParseBytes(requireValue("--max-bytes"));
+        } else if (arg == "--batch-counts") {
+            options.batchCounts = ParseList(requireValue("--batch-counts"), false);
+        } else if (arg == "--batch-chunks") {
+            options.batchChunkBytes = ParseList(requireValue("--batch-chunks"), true);
+        } else {
+            throw std::invalid_argument("unknown argument: " + arg);
+        }
+    }
+
+    if (options.iterations == 0) { throw std::invalid_argument("--iters must be greater than 0"); }
+    if (options.minBytes == 0 || options.maxBytes == 0 || options.minBytes > options.maxBytes) {
+        throw std::invalid_argument("invalid min/max bytes");
+    }
+    return options;
+}
+
+}  // namespace UC::Transport::Ffts::Bench

ucm/transport/p2p/ffts/bench/ffts_bench_options.h

@@ -0,0 +1,58 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2026 Huawei Technologies Co., Ltd. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ * */
+#ifndef UNIFIEDCACHE_TRANSPORT_FFTS_BENCH_OPTIONS_H
+#define UNIFIEDCACHE_TRANSPORT_FFTS_BENCH_OPTIONS_H
+
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
+namespace UC::Transport::Ffts::Bench {
+
+constexpr size_t kKiB = 1024ULL;
+constexpr size_t kMiB = 1024ULL * kKiB;
+constexpr size_t kGiB = 1024ULL * kMiB;
+
+enum class Scenario {
+    All,
+    Single,
+    Batch,
+};
+
+struct Options {
+    int32_t deviceId{0};
+    size_t warmup{10};
+    size_t iterations{100};
+    size_t minBytes{4ULL * kKiB};
+    size_t maxBytes{256ULL * kMiB};
+    Scenario scenario{Scenario::All};
+    std::vector<size_t> batchCounts{4, 16, 64, 128};
+    std::vector<size_t> batchChunkBytes{4ULL * kKiB, 16ULL * kKiB, 64ULL * kKiB, 256ULL * kKiB, 1ULL * kMiB};
+};
+
+Options ParseOptions(int argc, char** argv);
+
+}  // namespace UC::Transport::Ffts::Bench
+
+#endif  // UNIFIEDCACHE_TRANSPORT_FFTS_BENCH_OPTIONS_H