diff --git a/CMakeLists.txt b/CMakeLists.txt index 41b3772..de47ddf 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -108,11 +108,13 @@ endif() # The S3 benchmark needs the io library (rest backend) and the benchmark tree. # Checked after the io gating above so a force-disabled io also disables it. -if(CUCASCADE_BUILD_S3_BENCHMARK AND (NOT CUCASCADE_BUILD_IO - OR NOT CUCASCADE_BUILD_BENCHMARKS)) +if(CUCASCADE_BUILD_S3_BENCHMARK + AND (NOT CUCASCADE_BUILD_IO + OR NOT CUCASCADE_BUILD_BENCHMARKS + OR NOT CUCASCADE_BUILD_CUDF)) message( STATUS - "CUCASCADE_BUILD_S3_BENCHMARK disabled: requires CUCASCADE_BUILD_IO=ON and CUCASCADE_BUILD_BENCHMARKS=ON" + "CUCASCADE_BUILD_S3_BENCHMARK disabled: requires CUCASCADE_BUILD_IO=ON, CUCASCADE_BUILD_BENCHMARKS=ON and CUCASCADE_BUILD_CUDF=ON (it links cucascade_cudf and kvikIO)" ) set(CUCASCADE_BUILD_S3_BENCHMARK OFF @@ -158,10 +160,15 @@ if(NOT CUCASCADE_TOPOLOGY_ONLY) pkg_check_modules(CURL REQUIRED IMPORTED_TARGET libcurl) find_package(OpenSSL REQUIRED) - # kvikIO — backs the local-file fallback ioctx (kvikio_context). Used - # directly (not via cudf) so the io library stays cudf-free. Not swappable: - # unlike moodycamel/invocable below there is no in-tree stand-in to replace. - find_package(kvikio REQUIRED CONFIG) + # kvikIO — backs the local-file fallback ioctx (kvikio_context). It reaches + # the environment only through libcudf's dependency closure (libkvikio is + # not a direct dependency), so it is tied to CUCASCADE_BUILD_CUDF: a + # cudf-free build drops kvikio_context and its catch-all registry entry, + # leaving uring/restful to claim paths. Consumers see the difference via the + # CUCASCADE_HAS_KVIKIO definition propagated by cucascade_io_thirdparty. + if(CUCASCADE_BUILD_CUDF) + find_package(kvikio REQUIRED CONFIG) + endif() # cucascade_io_thirdparty carries the swappable moodycamel + invocable # (abseil) usage requirements from a single place; the io object library, @@ -239,6 +246,15 @@ if(NOT CUCASCADE_TOPOLOGY_ONLY) target_compile_definitions(cucascade_io_thirdparty INTERFACE CUCASCADE_USE_ABSEIL_INVOCABLE) endif() + + # Gates the kvikIO fallback ioctx in io_config and the datasource registry. + # Carried on the same INTERFACE target as the definitions above so the io + # object library, its installable variants, and installed consumers all + # agree on the layout of io_config. + if(CUCASCADE_BUILD_CUDF) + target_compile_definitions(cucascade_io_thirdparty + INTERFACE CUCASCADE_HAS_KVIKIO) + endif() endif() # Find numa (provided by numactl-devel or libnuma-dev depending on the package @@ -407,7 +423,10 @@ if(CUCASCADE_BUILD_IO) # side by cuCascadeConfig.cmake (same names), mirroring the Numa::Numa # approach. set(CUCASCADE_IO_LINK_LIBS PkgConfig::LIBURING PkgConfig::CURL - OpenSSL::Crypto kvikio::kvikio) + OpenSSL::Crypto) + if(CUCASCADE_BUILD_CUDF) + list(APPEND CUCASCADE_IO_LINK_LIBS kvikio::kvikio) + endif() target_link_libraries( cucascade_io_objects PUBLIC cucascade_objects ${CUCASCADE_IO_LINK_LIBS} @@ -425,7 +444,7 @@ endif() target_include_directories(cucascade_topology_discovery_objects PUBLIC ${CUCASCADE_PUBLIC_INCLUDE_DIRS}) target_link_libraries(cucascade_topology_discovery_objects - PUBLIC CUDA::nvml_static rmm::rmm) + PUBLIC CUDA::nvml_static ${CMAKE_DL_LIBS}) target_compile_features(cucascade_topology_discovery_objects PUBLIC cxx_std_20) target_compile_features(cucascade_topology_discovery_objects PRIVATE cuda_std_20) @@ -443,7 +462,7 @@ if(CUCASCADE_BUILD_STATIC_LIBS) cucascade_topology_discovery_static) target_link_libraries(cucascade_topology_discovery_static - PRIVATE CUDA::nvml_static rmm::rmm) + PRIVATE CUDA::nvml_static ${CMAKE_DL_LIBS}) target_include_directories(cucascade_topology_discovery_static PUBLIC ${CUCASCADE_PUBLIC_INCLUDE_DIRS}) target_compile_features(cucascade_topology_discovery_static PUBLIC cxx_std_20) @@ -521,7 +540,7 @@ if(CUCASCADE_BUILD_SHARED_LIBS) cucascade_topology_discovery_shared) target_link_libraries(cucascade_topology_discovery_shared - PRIVATE CUDA::nvml_static rmm::rmm) + PRIVATE CUDA::nvml_static ${CMAKE_DL_LIBS}) target_include_directories(cucascade_topology_discovery_shared PUBLIC ${CUCASCADE_PUBLIC_INCLUDE_DIRS}) target_compile_features(cucascade_topology_discovery_shared PUBLIC cxx_std_20) diff --git a/include/cucascade/io/config.hpp b/include/cucascade/io/config.hpp index 250120b..f1b125c 100644 --- a/include/cucascade/io/config.hpp +++ b/include/cucascade/io/config.hpp @@ -18,7 +18,9 @@ #pragma once #include +#ifdef CUCASCADE_HAS_KVIKIO #include +#endif #include #include #include @@ -35,7 +37,9 @@ namespace cucascade::io { * Sub-configs: * - @c local — uring reactor tunables (local-disk IO path). * - @c rest — REST reactor tunables (S3/object-store IO path). - * - @c kvikio — kvikIO fallback tunables (local-disk catch-all path). + * - @c kvikio — kvikIO fallback tunables (local-disk catch-all path); present + * only when the library is built with CUCASCADE_BUILD_CUDF, which is what + * supplies kvikIO. * - @c cache — prefetching cache tunables. * - @c object_store — object-store credentials and endpoint. */ @@ -59,11 +63,13 @@ struct io_config { /// retry policy, etc. rest::config rest{}; +#ifdef CUCASCADE_HAS_KVIKIO /// kvikIO fallback configuration — thread-pool size, task/bounce sizing, /// O_DIRECT, compat mode. All fields default to "unset", leaving kvikIO's /// own env-var-seeded defaults in place. Note these are process-global once /// applied; see @ref kvikio_config. kvikio_config kvikio{}; +#endif /// Prefetching cache configuration — in-flight budget, pool sizing, /// dispose-after-use policy. diff --git a/src/io/CMakeLists.txt b/src/io/CMakeLists.txt index d645721..f8a7a4e 100644 --- a/src/io/CMakeLists.txt +++ b/src/io/CMakeLists.txt @@ -25,10 +25,16 @@ target_sources( ${CMAKE_CURRENT_SOURCE_DIR}/rest/rest_reactor.cpp ${CMAKE_CURRENT_SOURCE_DIR}/uring/uring_ioctx.cpp ${CMAKE_CURRENT_SOURCE_DIR}/uring/uring_reactor.cpp - ${CMAKE_CURRENT_SOURCE_DIR}/kvikio/kvikio_context.cpp ${CMAKE_CURRENT_SOURCE_DIR}/rest/s3/sigv4.cpp ${CMAKE_CURRENT_SOURCE_DIR}/rest/s3/sigv4_authorizer.cpp ${CMAKE_CURRENT_SOURCE_DIR}/rest/s3/list_parser.cpp ${CMAKE_CURRENT_SOURCE_DIR}/cache/types.cpp ${CMAKE_CURRENT_SOURCE_DIR}/cache/metadata_store.cpp ${CMAKE_CURRENT_SOURCE_DIR}/cache/prefetching_cache.cpp) + +# kvikIO reaches the environment only via libcudf's dependency closure, so the +# fallback ioctx is built only alongside the cudf layer. +if(CUCASCADE_BUILD_CUDF) + target_sources(cucascade_io_objects + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/kvikio/kvikio_context.cpp) +endif() diff --git a/src/io/datasource_factory.cpp b/src/io/datasource_factory.cpp index edefff4..cf61c69 100644 --- a/src/io/datasource_factory.cpp +++ b/src/io/datasource_factory.cpp @@ -19,7 +19,9 @@ #include #include #include +#ifdef CUCASCADE_HAS_KVIKIO #include +#endif #include #include #include @@ -78,6 +80,7 @@ std::shared_ptr make_s3_authorizer(const object_store_ using scheme_checker_type = io_context_registry::scheme_checker_type; using factory_type = io_context_registry::factory_type; +#ifdef CUCASCADE_HAS_KVIKIO factory_type make_kvikio_ioctx_factory() { return [](const io_config& config) -> std::shared_ptr { @@ -91,6 +94,7 @@ factory_type make_kvikio_ioctx_factory() } }; } +#endif factory_type make_uring_ioctx_factory( cucascade::memory::memory_reservation_manager& reservation_manager) @@ -160,11 +164,14 @@ io_context_registry::io_context_registry( // uring / rest claim paths via their reactor's static supports() (local // files and s3:// URLs respectively). kvikio is the universal fallback — // it can open any local path — so it matches everything and lookup_path - // defers it behind the explicit backends. + // defers it behind the explicit backends. Without kvikIO (a cudf-free + // build) there is no catch-all and unmatched paths resolve to nothing. +#ifdef CUCASCADE_HAS_KVIKIO _entries.emplace( io_context_type::kvikio, entry{ io_context_type::kvikio, [](std::string_view) { return true; }, make_kvikio_ioctx_factory()}); +#endif _entries.emplace(io_context_type::uring, entry{io_context_type::uring, &uring::uring_reactor::supports, diff --git a/src/memory/topology_discovery.cpp b/src/memory/topology_discovery.cpp index a8f303f..12bf6fa 100644 --- a/src/memory/topology_discovery.cpp +++ b/src/memory/topology_discovery.cpp @@ -5,8 +5,7 @@ #include -#include -#include +#include #include #include @@ -61,25 +60,108 @@ void report_nvml_error(nvmlReturn_t result, std::string const& context) } /** - * @brief Query whether a CUDA device supports hardware-accelerated decompression. + * @brief Minimal subset of the CUDA driver API used for device capability queries. * - * Delegates to `rmm::detail::hwdecompress::is_supported()`, which checks the CUDA - * driver version. RMM's capability queries are scoped to the current device, so the - * call is wrapped in an `rmm::cuda_set_device_raii`. Best-effort: any failure while - * setting the device or probing yields false. + * Resolved with dlopen/dlsym rather than linked, so this library keeps loading on + * hosts without an NVIDIA driver, matching how NVML is treated here. + */ +struct cuda_driver_api { + CUresult (*init)(unsigned int){nullptr}; + CUresult (*device_get_by_pci_bus_id)(CUdevice*, char const*){nullptr}; + CUresult (*device_get_attribute)(int*, CUdevice_attribute, CUdevice){nullptr}; + bool available{false}; +}; + +/** + * @brief Resolve a symbol from an already-opened shared object. * - * @param cuda_ordinal CUDA device ordinal (matches the runtime device index used - * elsewhere in discovery under the same CUDA_VISIBLE_DEVICES ordering). - * @return true iff the hardware decompression engine is available. + * A null return from `dlsym` is not by itself an error — a symbol may legitimately + * have a null value — so failure is detected by clearing `dlerror()` beforehand and + * inspecting it afterwards. + * + * @tparam Fn Function pointer type of the symbol. + * @param fn Set to the resolved symbol on success; left untouched on failure. + * @param handle Handle returned by `dlopen`. + * @param name Symbol name. + * @return true if the symbol was resolved. */ -bool query_hw_decompression(unsigned int cuda_ordinal) +template +bool load_symbol(Fn& fn, void* handle, char const* name) { - try { - rmm::cuda_set_device_raii set_device{rmm::cuda_device_id{static_cast(cuda_ordinal)}}; - return rmm::detail::hwdecompress::is_supported(); - } catch (...) { + ::dlerror(); + auto* symbol = reinterpret_cast(dlsym(handle, name)); + if (::dlerror() != nullptr) { return false; } + fn = symbol; + return true; +} + +/** + * @brief Load and initialize the CUDA driver API once per process. + * + * The library handle is intentionally never `dlclose`d — it is held for the process + * lifetime, mirroring the init-once treatment of NVML in `discover()`. + * + * @return The resolved entry points; `available` is false if the driver is missing, + * a symbol could not be resolved, or `cuInit` failed. + */ +cuda_driver_api const& load_cuda_driver_api() +{ + static cuda_driver_api const api = [] { + cuda_driver_api resolved; + + void* handle = dlopen("libcuda.so.1", RTLD_LAZY | RTLD_LOCAL); + if (handle == nullptr) { return resolved; } + + if (!load_symbol(resolved.init, handle, "cuInit") || + !load_symbol(resolved.device_get_by_pci_bus_id, handle, "cuDeviceGetByPCIBusId") || + !load_symbol(resolved.device_get_attribute, handle, "cuDeviceGetAttribute")) { + return cuda_driver_api{}; + } + if (resolved.init(0) != CUDA_SUCCESS) { return cuda_driver_api{}; } + + resolved.available = true; + return resolved; + }(); + return api; +} + +/** + * @brief Query whether a GPU has a hardware-accelerated decompression engine. + * + * The device is identified by PCI bus id rather than by ordinal. Device ordinals are + * not a stable identity across APIs: NVML enumerates in PCI-bus order while the CUDA + * runtime defaults to `CUDA_DEVICE_ORDER=FASTEST_FIRST`, so the index of a GPU in + * this discovery's list need not name the same device to CUDA on a heterogeneous + * host. `cuDeviceGetByPCIBusId` sidesteps both that reordering and any + * `CUDA_VISIBLE_DEVICES` remapping. + * + * `cuDeviceGetAttribute` takes its device explicitly, so no context is created and + * the calling thread's current device is left untouched. + * + * Best-effort: a missing driver, a bus id CUDA does not expose (e.g. masked out by + * `CUDA_VISIBLE_DEVICES`), or an attribute unsupported by the running driver all + * yield false. + * + * @param pci_bus_id PCI bus id of the GPU, in NVML's `domain:bus:device.function` + * form. For a MIG instance this is the parent physical GPU's bus id, which is the + * correct scope: the decompression engine is a property of the physical device. + * @return true iff the device reports at least one hardware decompression algorithm. + */ +bool query_hw_decompression(std::string const& pci_bus_id) +{ + auto const& api = load_cuda_driver_api(); + if (!api.available || pci_bus_id.empty()) { return false; } + + CUdevice device = 0; + if (api.device_get_by_pci_bus_id(&device, pci_bus_id.c_str()) != CUDA_SUCCESS) { return false; } + + int algorithm_mask = 0; + if (api.device_get_attribute(&algorithm_mask, + CU_DEVICE_ATTRIBUTE_MEM_DECOMPRESS_ALGORITHM_MASK, + device) != CUDA_SUCCESS) { return false; } + return algorithm_mask != 0; } /** @@ -99,7 +181,6 @@ std::string read_file_content(std::string const& path) std::stringstream buffer; buffer << file.rdbuf(); std::string content = buffer.str(); - // Trim trailing newline if (!content.empty() && content.back() == '\n') { content.pop_back(); } return content; } @@ -125,14 +206,12 @@ std::vector parse_cpu_list(std::string const& cpulist) while (std::getline(iss, token, ',')) { size_t dash_pos = token.find('-'); if (dash_pos != std::string::npos) { - // Range, e.g., "0-31" int start = std::stoi(token.substr(0, dash_pos)); int end = std::stoi(token.substr(dash_pos + 1)); for (int i = start; i <= end; ++i) { cores.push_back(i); } } else { - // Single core, e.g., "5" cores.push_back(std::stoi(token)); } } @@ -158,7 +237,6 @@ std::string normalize_pci_bus_id(std::string const& pci_bus_id) std::string domain = pci_bus_id.substr(0, colon_pos); if (domain.length() > 4) { domain = domain.substr(domain.length() - 4); } - // Convert to lowercase std::string normalized_id = domain + pci_bus_id.substr(colon_pos); std::ranges::transform(normalized_id, normalized_id.begin(), ::tolower); @@ -381,7 +459,6 @@ PciePathType get_pcie_path_type(std::string const& gpu_pci_id, std::string const std::string gpu_norm = normalize_pci_bus_id(gpu_pci_id); std::string nic_norm = normalize_pci_bus_id(nic_pci_id); - // Read NUMA nodes int gpu_numa = -1, nic_numa = -1; std::string gpu_numa_str = read_file_content("/sys/bus/pci/devices/" + gpu_norm + "/numa_node"); std::string nic_numa_str = read_file_content("/sys/bus/pci/devices/" + nic_norm + "/numa_node"); @@ -389,7 +466,6 @@ PciePathType get_pcie_path_type(std::string const& gpu_pci_id, std::string const if (!gpu_numa_str.empty()) { gpu_numa = std::stoi(gpu_numa_str); } if (!nic_numa_str.empty()) { nic_numa = std::stoi(nic_numa_str); } - // If different NUMA nodes, it's a SYS connection if (gpu_numa != nic_numa && gpu_numa >= 0 && nic_numa >= 0) { return PciePathType::SYS; } // Use PCI bus number proximity as a heuristic for connection quality @@ -564,7 +640,6 @@ std::vector discover_network_devices_with_topology( NetworkDeviceWithTopology dev; dev.name = entry.path().filename().string(); - // Get device's NUMA node and PCI bus ID std::string numa_path = entry.path().string() + "/device/numa_node"; std::string numa_str = read_file_content(numa_path); dev.numa_node = numa_str.empty() ? -1 : std::stoi(numa_str); @@ -594,7 +669,6 @@ std::vector discover_storage_devices_with_topology() dev.name = entry.path().filename().string(); dev.type = StorageDriveType::NVME; - // Get device's NUMA node and PCI bus ID std::string numa_path = entry.path().string() + "/device/numa_node"; std::string numa_str = read_file_content(numa_path); dev.numa_node = numa_str.empty() ? -1 : std::stoi(numa_str); @@ -628,7 +702,6 @@ std::vector map_network_devices_to_gpu( { std::vector mapped_devices; - // Structure to hold NIC with its topology path type struct NicWithPath { std::string name; PciePathType path_type; @@ -636,7 +709,6 @@ std::vector map_network_devices_to_gpu( std::vector nics_with_paths; - // Query topology distance for each NIC for (auto const& dev : network_devices) { if (dev.pci_bus_id.empty()) { continue; // Skip devices without PCI info @@ -649,7 +721,6 @@ std::vector map_network_devices_to_gpu( nics_with_paths.push_back(nic); } - // Find the best (lowest) path type if (nics_with_paths.empty()) { return mapped_devices; } PciePathType best_path_type = PciePathType::SYS; @@ -657,19 +728,16 @@ std::vector map_network_devices_to_gpu( if (nic.path_type < best_path_type) { best_path_type = nic.path_type; } } - // Return all NICs with the best path type for (auto const& nic : nics_with_paths) { if (nic.path_type == best_path_type) { mapped_devices.push_back(nic.name); } } - // If no devices found, fall back to NUMA-based mapping if (mapped_devices.empty()) { for (auto const& dev : network_devices) { if (dev.numa_node == gpu_numa_node) { mapped_devices.push_back(dev.name); } } } - // Last resort: return all devices if (mapped_devices.empty() && !network_devices.empty()) { for (auto const& dev : network_devices) { mapped_devices.push_back(dev.name); @@ -761,7 +829,6 @@ bool topology_discovery::discover(NetworkDeviceVerification net_verification) // Continue anyway to report system info even without GPUs } - // Get GPU count unsigned int device_count = 0; bool nvml_available = false; if (result == NVML_SUCCESS) { @@ -774,17 +841,14 @@ bool topology_discovery::discover(NetworkDeviceVerification net_verification) } } - // Discover network devices std::vector network_devices_with_topology = discover_network_devices_with_topology(net_verification); - // Get system information topology.hostname = get_hostname(); topology.num_numa_nodes = count_numa_nodes(); topology.num_gpus = device_count; topology.num_network_devices = static_cast(network_devices_with_topology.size()); - // Convert network devices to public format topology.network_devices.clear(); for (auto const& dev : network_devices_with_topology) { network_device_info info; @@ -796,7 +860,6 @@ bool topology_discovery::discover(NetworkDeviceVerification net_verification) topology.storage_devices = discover_storage_devices_with_topology(); - // Collect GPU information topology.gpus.clear(); std::vector nvml_gpus; @@ -909,7 +972,7 @@ bool topology_discovery::discover(NetworkDeviceVerification net_verification) if (nvml_idx >= nvml_gpus.size()) { continue; } auto gpu = nvml_gpus[nvml_idx]; gpu.id = static_cast(visible_idx); - gpu.hw_decompression_available = query_hw_decompression(gpu.id); + gpu.hw_decompression_available = query_hw_decompression(gpu.pci_bus_id); topology.gpus.push_back(std::move(gpu)); } diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 03de372..cef7d22 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -62,13 +62,18 @@ if(NOT CUCASCADE_TOPOLOGY_ONLY AND CUCASCADE_BUILD_IO) cucascade_io_tests io/test_uri_parser.cpp io/cache/test_metadata_store.cpp - io/kvikio/test_kvikio_config.cpp io/rest/test_shared_byte_span.cpp io/rest/s3/test_sigv4.cpp io/rest/s3/test_sigv4_authorizer.cpp io/rest/s3/test_static_credentials.cpp # Main test runner unittest.cpp) + + # kvikIO-backed sources exist only in a cudf build; see src/io/CMakeLists.txt. + if(CUCASCADE_BUILD_CUDF) + target_sources(cucascade_io_tests PRIVATE io/kvikio/test_kvikio_config.cpp) + endif() + set_target_properties(cucascade_io_tests PROPERTIES CUDA_STANDARD 20 CUDA_STANDARD_REQUIRED ON)