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

feat: add opt-in volatile memory chunk storage for IO500

parent baa4066a
Loading
Loading
Loading
Loading
Loading

docs/memory_storage.md

0 → 100644
+29 −0
Changes for docs/memory_storage.md: 29 added lines, 0 removed lines.
Original line number Diff line number Diff line
# Experimental volatile chunk storage for IO500

Set `GKFS_USE_MEM=ON` and `GKFS_MEM_CAPACITY_GB=900` on every daemon.
The capacity is **decimal GB**, per daemon, for chunk mappings only. The default
remains filesystem-backed storage when `GKFS_USE_MEM` is unset or OFF.

A single private anonymous mmap arena reserves virtual address space; physical
pages are allocated on demand. Deleted chunks return pages with madvise. They do not occupy
`/dev/shm` and disappear when the daemon exits. Each chunk reserves one full
configured chunk size against the budget; untouched pages remain demand-zero.
The budget excludes metadata, inline payloads, map bookkeeping, RPC buffers and
client memory. Capacity exhaustion returns ENOSPC; mmap failure returns its
system error. Linux can still OOM-kill a process if physical/job memory is
exhausted while faulting pages: a configured budget is not a RAM reservation.

Metadata and inline small files still use RocksDB under `--metadir` (rootdir
by default). Put the job-specific root/metadir on `/dev/shm/rnou` to retain
inline data in tmpfs. This consumes the same physical RAM as anonymous chunks;
reserve room for both. No new persistence guarantee is provided by fsync:
the existing client fsync waits for async writes and flushes size metadata.

The hostfile advertises `storage=memory`; matching clients automatically avoid
the direct local-chunk-file shortcut, even without inheriting GKFS_USE_MEM.
Mixed memory/disk deployments are rejected. Use matching daemon/client builds;
old clients do not understand this marker. CAS/two-tier CAS, expansion,
replication and mutation are unsupported. Disk calibration/space weights are
skipped, using equal placement weights. Normal metadata optimizations and
inline files remain available. This mode is intended for fresh IO500 instances,
not restart recovery or persistent production data.
 No newline at end of file
+2 −0
Changes for include/common/env.hpp: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -84,6 +84,8 @@ static constexpr auto USE_DIRENTS_COMPRESSION =
static constexpr auto ENABLE_TRACE = ADD_PREFIX("ENABLE_TRACE");
static constexpr auto TRACE_SAMPLE_RATE = ADD_PREFIX("TRACE_SAMPLE_RATE");
static constexpr auto ENABLE_CAS = ADD_PREFIX("ENABLE_CAS");
static constexpr auto USE_MEM = ADD_PREFIX("USE_MEM");
static constexpr auto MEM_CAPACITY_GB = ADD_PREFIX("MEM_CAPACITY_GB");
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");
+3 −0
Changes for include/daemon/backend/data/chunk_storage.hpp: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -60,6 +60,8 @@ class logger;

namespace gkfs::data {

class MemoryStorage;

struct ChunkStat {
    unsigned long chunk_size;
    unsigned long chunk_total;
@@ -88,6 +90,7 @@ private:
    bool cas_enabled_{false};
    size_t cas_min_size_{16384};
    std::string cas_dir_;
    std::shared_ptr<MemoryStorage> memory_;

    /**
     * @brief Converts an internal gkfs path under the root dir to the absolute
+202 −0
Changes for include/daemon/backend/data/memory_storage.hpp: 202 added lines, 0 removed lines.
Original line number Diff line number Diff line
/* Copyright 2026 Barcelona Supercomputing Center (BSC), Spain
 * SPDX-License-Identifier: GPL-3.0-or-later
 */

#ifndef GKFS_MEMORY_STORAGE_HPP
#define GKFS_MEMORY_STORAGE_HPP

#include <daemon/backend/data/chunk_storage.hpp>
#include <algorithm>
#include <array>
#include <atomic>
#include <cerrno>
#include <cstring>
#include <mutex>
#include <unordered_map>
#include <vector>
#include <sys/mman.h>
#include <unistd.h>

namespace gkfs::data {

// Volatile IO500 backend. Sharding allows unrelated files to transfer in
// parallel.
class MemoryStorage {
    char* arena_{nullptr};
    std::mutex arena_mutex_;
    std::vector<size_t> free_slots_;
    size_t next_slot_{0};

    void
    release(char* data) {
        // Return physical pages without splitting the arena into millions of
        // VMAs.
        if(madvise(data, chunk_size_, MADV_DONTNEED) != 0)
            std::memset(data, 0, chunk_size_);
        std::lock_guard<std::mutex> lock(arena_mutex_);
        free_slots_.push_back(static_cast<size_t>(data - arena_) / chunk_size_);
    }

    struct Chunk {
        char* data;
        size_t length{0};
        MemoryStorage& owner;
        explicit Chunk(MemoryStorage& store) : owner(store) {
            std::lock_guard<std::mutex> lock(store.arena_mutex_);
            size_t slot;
            if(store.free_slots_.empty()) {
                slot = store.next_slot_++;
            } else {
                slot = store.free_slots_.back();
                store.free_slots_.pop_back();
            }
            data = store.arena_ + slot * store.chunk_size_;
        }
        ~Chunk() {
            owner.release(data);
        }
        Chunk(const Chunk&) = delete;
        Chunk&
        operator=(const Chunk&) = delete;
    };
    using Chunks =
            std::unordered_map<gkfs::rpc::chnk_id_t, std::unique_ptr<Chunk>>;
    struct Shard {
        std::mutex mutex;
        std::unordered_map<std::string, Chunks> files;
    };
    std::array<Shard, 256> shards_;
    const size_t chunk_size_;
    const size_t capacity_;
    std::atomic<size_t> used_{0};

    Shard&
    shard(const std::string& path) {
        return shards_[std::hash<std::string>{}(path) % shards_.size()];
    }
    void
    reserve() {
        auto used = used_.load();
        do {
            if(chunk_size_ > capacity_ - used)
                throw ChunkStorageException(ENOSPC,
                                            "Memory data budget exhausted");
        } while(!used_.compare_exchange_weak(used, used + chunk_size_));
    }
    void
    check(size_t size, off64_t offset) const {
        if(offset < 0 || static_cast<size_t>(offset) > chunk_size_ ||
           size > chunk_size_ - static_cast<size_t>(offset))
            throw ChunkStorageException(EINVAL, "Invalid memory chunk range");
    }

public:
    MemoryStorage(size_t chunk_size, size_t capacity)
        : chunk_size_(chunk_size), capacity_(capacity) {
        if(!chunk_size || capacity < chunk_size)
            throw ChunkStorageException(EINVAL,
                                        "Invalid memory storage capacity");
        if(chunk_size % static_cast<size_t>(sysconf(_SC_PAGESIZE)) != 0)
            throw ChunkStorageException(EINVAL,
                                        "Memory chunks must be page aligned");
        free_slots_.reserve(capacity / chunk_size);
        void* p = mmap(nullptr, capacity, PROT_READ | PROT_WRITE,
                       MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
        if(p == MAP_FAILED)
            throw ChunkStorageException(errno, "Memory arena mmap failed");
        arena_ = static_cast<char*>(p);
    }
    ~MemoryStorage() {
        for(auto& s : shards_) s.files.clear();
        munmap(arena_, capacity_);
    }
    ssize_t
    write(const std::string& path, gkfs::rpc::chnk_id_t id, const char* buf,
          size_t size, off64_t offset, bool replace = false) {
        check(size, offset);
        auto& s = shard(path);
        std::lock_guard<std::mutex> lock(s.mutex);
        auto& chunks = s.files[path];
        auto it = chunks.find(id);
        if(it == chunks.end()) {
            bool reserved = false;
            try {
                reserve();
                reserved = true;
                auto chunk = std::make_unique<Chunk>(*this);
                it = chunks.emplace(id, std::move(chunk)).first;
            } catch(...) {
                if(reserved)
                    used_.fetch_sub(chunk_size_);
                if(chunks.empty())
                    s.files.erase(path);
                throw;
            }
        }
        auto& c = *it->second;
        if(replace) {
            // Keep later sparse extension zero-filled after replacement.
            std::memset(c.data, 0, c.length);
            c.length = 0;
        }
        if(size)
            std::memcpy(c.data + offset, buf, size);
        c.length = std::max(c.length, static_cast<size_t>(offset) + size);
        return size;
    }
    ssize_t
    read(const std::string& path, gkfs::rpc::chnk_id_t id, char* buf,
         size_t size, off64_t offset) {
        check(size, offset);
        auto& s = shard(path);
        std::lock_guard<std::mutex> lock(s.mutex);
        auto f = s.files.find(path);
        if(f == s.files.end() || !f->second.count(id))
            throw ChunkStorageException(ENOENT, "Memory chunk missing");
        auto& c = *f->second.at(id);
        if(static_cast<size_t>(offset) >= c.length)
            return 0;
        auto n = std::min(size, c.length - static_cast<size_t>(offset));
        if(n)
            std::memcpy(buf, c.data + offset, n);
        return n;
    }
    void
    trim(const std::string& path, gkfs::rpc::chnk_id_t start) {
        auto& s = shard(path);
        std::lock_guard<std::mutex> lock(s.mutex);
        auto f = s.files.find(path);
        if(f == s.files.end())
            return;
        for(auto it = f->second.begin(); it != f->second.end();) {
            if(it->first >= start) {
                it = f->second.erase(it);
                used_.fetch_sub(chunk_size_);
            } else {
                ++it;
            }
        }
        if(f->second.empty())
            s.files.erase(f);
    }
    void
    truncate(const std::string& path, gkfs::rpc::chnk_id_t id, off_t length) {
        check(0, length);
        auto& s = shard(path);
        std::lock_guard<std::mutex> lock(s.mutex);
        auto f = s.files.find(path);
        if(f == s.files.end() || !f->second.count(id))
            return;
        auto& c = *f->second.at(id);
        if(static_cast<size_t>(length) < c.length)
            std::memset(c.data + length, 0, c.length - length);
        c.length = length;
    }
    ChunkStat
    stat() const {
        return {chunk_size_, capacity_ / chunk_size_,
                (capacity_ - used_.load()) / chunk_size_};
    }
};
} // namespace gkfs::data
#endif
 No newline at end of file
+4 −0
Changes for src/client/preload.cpp: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -785,6 +785,10 @@ init_preload() {
    } else if(gkfs::env::var_is_set("LIBGKFS_LOCAL_BYPASS")) {
        bypass_enabled = gkfs::env::get_var_bool("LIBGKFS_LOCAL_BYPASS", true);
    }
    if(gkfs::env::get_var_bool(gkfs::env::USE_MEM, false) ||
       CTX->fs_conf()->rootdir.empty()) {
        bypass_enabled = false;
    }
    CTX->local_bypass(bypass_enabled);
    LOG(INFO, "Local chunk bypass: {}", CTX->local_bypass() ? "ON" : "OFF");

Loading