Commit a056db4b authored by Ramon Nou's avatar Ramon Nou
Browse files

bug on transfer

parent 9e8136e3

CMake/FindZStd.cmake

+37 −20
Changes for CMake/FindZStd.cmake: 37 added lines, 20 removed lines.
Original line number Diff line number Diff line
@@ -26,32 +26,49 @@ 
# SPDX-License-Identifier: GPL-3.0-or-later                                    #

################################################################################



#

# - Try to find Facebook zstd library

# This will define

# ZStd_FOUND

# ZStd_INCLUDE_DIR

# ZStd_LIBRARIES

#





# Standard names to search for

set(ZStd_NAMES zstd zstd_static)



find_path(ZStd_INCLUDE_DIR

          NAMES zstd.h

)

          PATH_SUFFIXES include)



find_library(ZStd_LIBRARY

    NAMES zstd

)

# Allow ZStd_LIBRARY to be set manually, as the location of the zstd library

if(NOT ZStd_LIBRARY)

  find_library(ZStd_LIBRARY_RELEASE

               NAMES ${ZStd_NAMES}

               PATH_SUFFIXES lib)

  

set(ZStd_LIBRARIES ${ZStd_LIBRARY})

set(ZStd_INCLUDE_DIRS ${ZStd_INCLUDE_DIR})



  include(SelectLibraryConfigurations)

  select_library_configurations(ZStd)

endif()



unset(ZStd_NAMES)



mark_as_advanced(ZStd_INCLUDE_DIR)



include(FindPackageHandleStandardArgs)

FIND_PACKAGE_HANDLE_STANDARD_ARGS(ZStd

        REQUIRED_VARS ZStd_LIBRARY ZStd_INCLUDE_DIR

        VERSION_VAR ZStd_VERSION_STRING)



if(ZStd_FOUND)

    set(ZStd_INCLUDE_DIRS ${ZStd_INCLUDE_DIR})



    if(NOT ZStd_LIBRARIES)

        set(ZStd_LIBRARIES ${ZStd_LIBRARY})

    endif()



    if(NOT TARGET ZStd::ZStd)

        add_library(ZStd::ZStd UNKNOWN IMPORTED)

        set_target_properties(ZStd::ZStd PROPERTIES

                INTERFACE_INCLUDE_DIRECTORIES "${ZStd_INCLUDE_DIRS}")

       

find_package_handle_standard_args(ZStd

    DEFAULT_MSG ZStd_LIBRARY ZStd_INCLUDE_DIR

)

        set_target_properties(ZStd::ZStd PROPERTIES

                IMPORTED_LOCATION "${ZStd_LIBRARY}")



mark_as_advanced(

    ZStd_LIBRARY

    ZStd_INCLUDE_DIR

)
 
 No newline at end of file
 
    endif()

endif()

examples/gfind/CMakeLists.txt

+1 −1
Changes for examples/gfind/CMakeLists.txt: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -29,7 +29,7 @@ 
set (CMAKE_CXX_STANDARD 17)



add_executable(sfind sfind.cpp)

target_link_libraries(sfind PRIVATE zstd)

target_link_libraries(sfind PRIVATE ZStd::ZStd)

set_property(TARGET sfind PROPERTY POSITION_INDEPENDENT_CODE ON)

if(GKFS_INSTALL_TESTS)

    install(TARGETS sfind

src/client/CMakeLists.txt

+1 −1
Changes for src/client/CMakeLists.txt: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -61,7 +61,7 @@ target_link_libraries( 
    fmt::fmt

    Threads::Threads

    Microsoft.GSL::GSL

    zstd

    ZStd::ZStd

)



add_library(gkfs_intercept SHARED)

src/daemon/CMakeLists.txt

+1 −1
Changes for src/daemon/CMakeLists.txt: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -73,7 +73,7 @@ target_link_libraries( 
    Margo::Margo

    # others

    Threads::Threads

    zstd

    ZStd::ZStd

)



if (GKFS_ENABLE_AGIOS)

src/daemon/handler/srv_metadata.cpp

+40 −18
Changes for src/daemon/handler/srv_metadata.cpp: 40 added lines, 18 removed lines.
Original line number Diff line number Diff line
@@ -727,8 +727,9 @@ rpc_srv_get_dirents_extended(hg_handle_t handle) { 
    auto hgi = margo_get_info(handle);

    auto mid = margo_hg_handle_get_instance(handle);

    auto client_bulk_size = margo_bulk_get_size(in.bulk_handle);

    GKFS_DATA->spdlogger()->debug("{}() Got RPC: path '{}' client_bulk_size '{}' ",

                                  __func__, in.path, client_bulk_size);

    GKFS_DATA->spdlogger()->debug(

            "{}() Got RPC: path '{}' client_bulk_size '{}' ", __func__, in.path,

            client_bulk_size);



    // Get directory entries from local DB (Unchanged)

    vector<tuple<string, bool, size_t, time_t>> entries{};
@@ -749,7 +750,8 @@ rpc_srv_get_dirents_extended(hg_handle_t handle) { 
    // === STEP 1: Serialize data into a temporary uncompressed buffer ===

    // This is simpler and safer than pre-calculating the exact size.

    std::vector<char> uncompressed_data;

    uncompressed_data.reserve(client_bulk_size); // Optimization: Reserve a reasonable starting size

    uncompressed_data.reserve(client_bulk_size); // Optimization: Reserve a

                                                 // reasonable starting size



    for(const auto& e : entries) {

        const auto& name = get<0>(e);
@@ -757,11 +759,22 @@ rpc_srv_get_dirents_extended(hg_handle_t handle) { 
        size_t file_size = get<2>(e);

        time_t ctime = get<3>(e);



        // Append data fields sequentially into the vector. The client will parse in this exact order.

        uncompressed_data.insert(uncompressed_data.end(), reinterpret_cast<const char*>(&is_dir), reinterpret_cast<const char*>(&is_dir) + sizeof(is_dir));

        uncompressed_data.insert(uncompressed_data.end(), reinterpret_cast<const char*>(&file_size), reinterpret_cast<const char*>(&file_size) + sizeof(file_size));

        uncompressed_data.insert(uncompressed_data.end(), reinterpret_cast<const char*>(&ctime), reinterpret_cast<const char*>(&ctime) + sizeof(ctime));

        uncompressed_data.insert(uncompressed_data.end(), name.c_str(), name.c_str() + name.length() + 1); // Append string and null terminator

        // Append data fields sequentially into the vector. The client will

        // parse in this exact order.

        uncompressed_data.insert(

                uncompressed_data.end(), reinterpret_cast<const char*>(&is_dir),

                reinterpret_cast<const char*>(&is_dir) + sizeof(is_dir));

        uncompressed_data.insert(uncompressed_data.end(),

                                 reinterpret_cast<const char*>(&file_size),

                                 reinterpret_cast<const char*>(&file_size) +

                                         sizeof(file_size));

        uncompressed_data.insert(

                uncompressed_data.end(), reinterpret_cast<const char*>(&ctime),

                reinterpret_cast<const char*>(&ctime) + sizeof(ctime));

        uncompressed_data.insert(

                uncompressed_data.end(), name.c_str(),

                name.c_str() + name.length() +

                        1); // Append string and null terminator

    }



    const size_t uncompressed_size = uncompressed_data.size();
@@ -770,14 +783,16 @@ rpc_srv_get_dirents_extended(hg_handle_t handle) { 
    const size_t compressed_bound = ZSTD_compressBound(uncompressed_size);

    std::vector<char> compressed_data(compressed_bound);



    const int compression_level = 3; // A good balance between speed and compression ratio

    const int compression_level =

            3; // A good balance between speed and compression ratio

    const size_t compressed_size = ZSTD_compress(

        compressed_data.data(), compressed_bound,

        uncompressed_data.data(), uncompressed_size,

        compression_level);

            compressed_data.data(), compressed_bound, uncompressed_data.data(),

            uncompressed_size, compression_level);



    if(ZSTD_isError(compressed_size)) {

        GKFS_DATA->spdlogger()->error("{}() Zstd compression failed: {}", __func__, ZSTD_getErrorName(compressed_size));

        GKFS_DATA->spdlogger()->error("{}() Zstd compression failed: {}",

                                      __func__,

                                      ZSTD_getErrorName(compressed_size));

        out.err = EIO;

        return gkfs::rpc::cleanup_respond(&handle, &in, &out);

    }
@@ -794,12 +809,16 @@ rpc_srv_get_dirents_extended(hg_handle_t handle) { 
    // === STEP 3: Transfer ONLY the compressed data ===

    void* bulk_buf;

    size_t transfer_size = compressed_size;

    ret = margo_bulk_create(mid, 1, &bulk_buf, &transfer_size, HG_BULK_WRITE_ONLY, &bulk_handle);

    ret = margo_bulk_create(mid, 1, &bulk_buf, &transfer_size,

                            HG_BULK_READ_ONLY, &bulk_handle);

    if(ret != HG_SUCCESS) {

        GKFS_DATA->spdlogger()->error("{}() Failed to create bulk handle for compressed data", __func__);

        GKFS_DATA->spdlogger()->error(

                "{}() Failed to create bulk handle for compressed data",

                __func__);

        return gkfs::rpc::cleanup_respond(&handle, &in, &out, &bulk_handle);

    }





    // A single, safe memcpy of the compressed data.

    memcpy(bulk_buf, compressed_data.data(), compressed_size);
@@ -807,15 +826,18 @@ rpc_srv_get_dirents_extended(hg_handle_t handle) { 
            "{}() Serialized '{}' entries to '{}' bytes, compressed to '{}' bytes. Transferring now.",

            __func__, entries.size(), uncompressed_size, compressed_size);



    ret = margo_bulk_transfer(mid, HG_BULK_PUSH, hgi->addr, in.bulk_handle, 0, bulk_handle, 0, compressed_size);

    ret = margo_bulk_transfer(mid, HG_BULK_PUSH, hgi->addr, in.bulk_handle, 0,

                              bulk_handle, 0, compressed_size);

    if(ret != HG_SUCCESS) {

        GKFS_DATA->spdlogger()->error("{}() Failed to push compressed data to client", __func__);

        GKFS_DATA->spdlogger()->error(

                "{}() Failed to push compressed data to client", __func__);

        out.err = EBUSY;

        return gkfs::rpc::cleanup_respond(&handle, &in, &out, &bulk_handle);

    }



    // === STEP 4: Respond to the client ===

    out.dirents_size = compressed_size; // This is the only size info the client needs from the struct

    out.dirents_size = compressed_size; // This is the only size info the client

                                        // needs from the struct

    out.err = 0;



    GKFS_DATA->spdlogger()->debug(