Commit 9f4387ed authored by Ramon Nou's avatar Ramon Nou
Browse files

feat(client): implement Two-Tier CAS shared memory architecture

parent 3a7aca10

include/client/preload_context.hpp

+39 −0
Changes for include/client/preload_context.hpp: 39 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -97,6 +97,10 @@ class WriteSizeCache; 
}

} // namespace cache



namespace shm {

class ShmCasManager;

} // namespace shm



namespace preload {

/*

 * Client file system config
@@ -196,6 +200,8 @@ private: 
    bool has_local_host_{false};

    bool cas_enabled_{false};

    size_t cas_min_size_{16384};

    bool two_tier_cas_enabled_{false};

    std::unique_ptr<gkfs::shm::ShmCasManager> shm_cas_;

    uint64_t stat_cache_ttl_ms_{500};

    mutable std::mutex stat_cache_mutex_;

    mutable std::unordered_map<std::string, StatCacheEntry> stat_cache_;
@@ -381,6 +387,39 @@ public: 
    void

    clear_stat_cache();



    bool

    two_tier_cas_enabled() const noexcept {

        return two_tier_cas_enabled_;

    }



    void

    two_tier_cas_enabled(bool enabled) noexcept {

        two_tier_cas_enabled_ = enabled;

    }



    bool

    init_two_tier_cas();



    void

    destroy_two_tier_cas();



    bool

    shm_cas_put(const std::string& path, mode_t mode, size_t size, time_t mtime,

                uint64_t cas_high, uint64_t cas_low,

                const std::string& inline_data);



    bool

    shm_cas_get(const std::string& path, mode_t& mode, size_t& size,

                time_t& mtime, uint64_t& cas_high, uint64_t& cas_low,

                std::string& inline_data);



    bool

    shm_cas_invalidate(const std::string& path);



    ssize_t

    shm_cas_read(const std::string& path, char* buf, size_t count,

                 off64_t offset);



    void

    clear_hosts();

include/client/shm_cas.hpp

0 → 100644
+125 −0
Changes for include/client/shm_cas.hpp: 125 added lines, 0 removed lines.
Original line number Diff line number Diff line 
/*

  Copyright 2018-2026, Barcelona Supercomputing Center (BSC), Spain

  Copyright 2015-2026, Johannes Gutenberg Universitaet Mainz, Germany



  This software was partially supported by the

  EC H2020 funded project NEXTGenIO (Project ID: 671951, www.nextgenio.eu).



  This software was partially supported by the

  ADA-FS project under the SPPEXA project funded by the DFG.



  This software was partially supported by the

  the European Union’s Horizon 2020 JTI-EuroHPC research and

  innovation programme, by the project ADMIRE (Project ID: 956748,

  admire-eurohpc.eu)



  This project was partially promoted by the Ministry for Digital Transformation

  and the Civil Service, within the framework of the Recovery,

  Transformation and Resilience Plan - Funded by the European Union

  -NextGenerationEU.



  This file is part of GekkoFS' POSIX interface.



  GekkoFS' POSIX interface is free software: you can redistribute it and/or

  modify it under the terms of the GNU Lesser General Public License as

  published by the Free Software Foundation, either version 3 of the License,

  or (at your option) any later version.



  GekkoFS' POSIX interface is distributed in the hope that it will be useful,

  but WITHOUT ANY WARRANTY; without even the implied warranty of

  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

  GNU Lesser General Public License for more details.



  You should have received a copy of the GNU Lesser General Public License

  along with GekkoFS' POSIX interface.  If not, see

  <https://www.gnu.org/licenses/>.



  SPDX-License-Identifier: LGPL-3.0-or-later

*/



#ifndef GKFS_CLIENT_SHM_CAS_HPP

#define GKFS_CLIENT_SHM_CAS_HPP



#include <cstdint>

#include <cstddef>

#include <string>

#include <atomic>

#include <sys/types.h>



namespace gkfs::shm {



static constexpr size_t SHM_CAS_MAX_INLINE = 2048;

static constexpr size_t SHM_CAS_MAX_PATH = 256;

static constexpr size_t SHM_CAS_WAYS = 4;

static constexpr size_t SHM_CAS_BUCKETS = 8192; // 32,768 entries total (~78 MB)



struct ShmCasEntry {

    std::atomic<uint32_t> seq{

            0};              // Seqlock: odd = write in progress, even = valid

    uint64_t path_hash{0};   // XXH3_64bits of full path

    uint64_t cas_high{0};    // Content hash (XXH3_128 high)

    uint64_t cas_low{0};     // Content hash (XXH3_128 low)

    uint32_t size{0};        // File size

    uint32_t mode{0};        // File mode

    uint32_t mtime{0};       // Modification timestamp

    uint32_t inline_size{0}; // Inline payload size (<= SHM_CAS_MAX_INLINE)

    char inline_payload[SHM_CAS_MAX_INLINE]{0};

    char path[SHM_CAS_MAX_PATH]{0};

};



struct ShmCasBucket {

    ShmCasEntry ways[SHM_CAS_WAYS];

    std::atomic<uint32_t> victim_idx{0};

};



struct ShmCasHeader {

    uint32_t magic;       // 0x474b4341 ('GKCA')

    uint32_t version;     // 1

    uint32_t num_buckets; // 8192

    uint32_t num_ways;    // 4

    uint64_t generation;  // Epoch / restart counter

};



class ShmCasManager {

private:

    bool initialized_{false};

    std::string shm_name_;

    void* mapped_region_{nullptr};

    size_t mapped_size_{0};

    ShmCasHeader* header_{nullptr};

    ShmCasBucket* buckets_{nullptr};



public:

    ShmCasManager() = default;

    ~ShmCasManager();



    bool

    init(uint64_t host_id);



    void

    destroy();



    bool

    put(const std::string& path, mode_t mode, size_t size, time_t mtime,

        uint64_t cas_high, uint64_t cas_low, const std::string& inline_data);



    bool

    get(const std::string& path, mode_t& mode, size_t& size, time_t& mtime,

        uint64_t& cas_high, uint64_t& cas_low, std::string& inline_data);



    bool

    invalidate(const std::string& path);



    ssize_t

    read_chunk(const std::string& rootdir, const std::string& path, char* buf,

               size_t count, off64_t offset);



    bool

    initialized() const noexcept {

        return initialized_;

    }

};



} // namespace gkfs::shm



#endif // GKFS_CLIENT_SHM_CAS_HPP

include/common/env.hpp

+1 −0
Changes for include/common/env.hpp: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -88,6 +88,7 @@ static constexpr auto CAS_MIN_SIZE = ADD_PREFIX("CAS_MIN_SIZE"); 
static constexpr auto LAZY_CONNECT = ADD_PREFIX("LAZY_CONNECT");

static constexpr auto STAT_CACHE_TTL_MS = ADD_PREFIX("STAT_CACHE_TTL_MS");

static constexpr auto NEGATIVE_STAT_CACHE = ADD_PREFIX("NEGATIVE_STAT_CACHE");

static constexpr auto ENABLE_TWO_TIER_CAS = ADD_PREFIX("ENABLE_TWO_TIER_CAS");



} // namespace gkfs::env

src/client/CMakeLists.txt

+1 −0
Changes for src/client/CMakeLists.txt: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -50,6 +50,7 @@ target_sources(gkfs_common 
    rpc/forward_metadata_proxy.cpp

    rpc/forward_malleability.cpp

    cache.cpp

    shm_cas.cpp

    syscalls/detail/syscall_info.c syscalls/util.S

    ../common/hostfile_management.cpp

    )

src/client/gkfs_data.cpp

+50 −0
Changes for src/client/gkfs_data.cpp: 50 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -50,6 +50,9 @@ 
#include <client/rpc/forward_data_proxy.hpp>

#include <client/open_dir.hpp>

#include <client/cache.hpp>

#include <client/shm_cas.hpp>

#define XXH_INLINE_ALL

#include <xxhash.h>

#include <string>

#include <string_view>

#include <filesystem>
@@ -140,8 +143,21 @@ gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count, 
                if(offset == 0) {

                    CTX->put_stat_cache(*path, md.serialize(),

                                        std::string(buf, count), true);

                    if(CTX->two_tier_cas_enabled()) {

                        XXH128_hash_t h = XXH3_128bits(buf, count);

                        std::string inline_str;

                        if(count <= gkfs::shm::SHM_CAS_MAX_INLINE) {

                            inline_str.assign(buf, count);

                        }

                        CTX->shm_cas_put(*path, file.mode() | S_IFREG, count,

                                         std::time(nullptr), h.high64, h.low64,

                                         inline_str);

                    }

                } else {

                    CTX->put_stat_cache(*path, md.serialize(), "", false);

                    if(CTX->two_tier_cas_enabled()) {

                        CTX->shm_cas_invalidate(*path);

                    }

                }

                if(CTX->use_dentry_cache() && CTX->dentry_cache()) {

                    std::filesystem::path p(*path);
@@ -206,6 +222,20 @@ gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count, 
                    file.path(), buf, offset, count, is_append, num_replicas);

            err = ret_inline.first;

            if(err == 0) {

                if(CTX->two_tier_cas_enabled()) {

                    if(offset == 0) {

                        XXH128_hash_t h = XXH3_128bits(buf, count);

                        std::string inline_str;

                        if(count <= gkfs::shm::SHM_CAS_MAX_INLINE) {

                            inline_str.assign(buf, count);

                        }

                        CTX->shm_cas_put(file.path(), file.mode() | S_IFREG,

                                         count, std::time(nullptr), h.high64,

                                         h.low64, inline_str);

                    } else {

                        CTX->shm_cas_invalidate(file.path());

                    }

                }

                if(update_pos)

                    file.pos(ret_inline.second + count);

                return count;
@@ -372,6 +402,19 @@ gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count, 
        errno = err;

        return -1;

    }

    if(CTX->two_tier_cas_enabled()) {

        if(offset == 0) {

            XXH128_hash_t h = XXH3_128bits(buf, count);

            std::string inline_str;

            if(count <= gkfs::shm::SHM_CAS_MAX_INLINE) {

                inline_str.assign(buf, count);

            }

            CTX->shm_cas_put(file.path(), file.mode() | S_IFREG, count,

                             std::time(nullptr), h.high64, h.low64, inline_str);

        } else {

            CTX->shm_cas_invalidate(file.path());

        }

    }

    if(update_pos) {

        // Update offset in file descriptor in the file map

        file.pos(offset + write_size);
@@ -550,6 +593,13 @@ gkfs_do_read(const gkfs::filemap::OpenFile& file, char* buf, size_t count, 
        memset(buf, 0, sizeof(char) * count);

    }



    if(CTX->two_tier_cas_enabled()) {

        ssize_t shm_r = CTX->shm_cas_read(file.path(), buf, count, offset);

        if(shm_r >= 0) {

            return shm_r;

        }

    }



    if(gkfs::config::metadata::use_inline_data && offset >= 0 &&

       static_cast<size_t>(offset) < gkfs::config::metadata::inline_data_size) {