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

feat(client): implement Option A lazy RPC initialization and clamp dirents initial budget

parent f266d6be

include/client/env.hpp

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

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

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

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

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

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

// Signal a live client watches for to reload its routing state (hosts +

include/client/preload.hpp

+4 −0
Changes for include/client/preload.hpp: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -54,6 +54,10 @@ void 
init_environment();

void

init_ld_env_if_needed();

void

ensure_environment_initialized();

bool

is_environment_initialized();

} // namespace gkfs::preload



void

src/client/gkfs_metadata.cpp

+4 −2
Changes for src/client/gkfs_metadata.cpp: 4 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -1317,11 +1317,13 @@ gkfs_opendir(const std::string& path) { 
        const auto targets =

                CTX->distributor()->locate_directory_metadata(path);

        const auto target_count = targets.size();

        const size_t per_target_budget =

                CTX->dirents_buff_size() / std::max<size_t>(size_t{1}, target_count);

        const bool use_proxy_dirents =

                gkfs::config::proxy::fwd_get_dirents_single && CTX->use_proxy();

        const size_t initial_buffer_size = std::clamp<size_t>(

                CTX->dirents_buff_size() / std::max<size_t>(size_t{1}, target_count),

                size_t{65536}, size_t{262144});

                per_target_budget, size_t{65536},

                std::max<size_t>(size_t{262144}, CTX->dirents_buff_size()));



        std::vector<std::vector<char>> buffers(target_count);

        std::vector<thallium::bulk> exposed_buffers;

src/client/path.cpp

+4 −2
Changes for src/client/path.cpp: 4 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -203,11 +203,13 @@ resolve(const string& path, bool resolve_last_link) { 
    }

    const auto& mountdir = CTX->mountdir();

    const bool is_mountdir =

            resolved == mountdir ||

            !mountdir.empty() &&

            (resolved == mountdir ||

             (resolved.size() > mountdir.size() &&

              resolved.compare(0, mountdir.size(), mountdir) == 0 &&

             resolved.at(mountdir.size()) == path::separator);

              resolved.at(mountdir.size()) == path::separator));

    if(is_mountdir) {

        gkfs::preload::ensure_environment_initialized();

        resolved.erase(1, mountdir.size());

        LOG(DEBUG, "internal: \"{}\"", resolved);

        return make_pair(true, resolved);

src/client/preload.cpp

+88 −40
Changes for src/client/preload.cpp: 88 added lines, 40 removed lines.
Original line number Diff line number Diff line
@@ -44,6 +44,7 @@ 
#include <client/path.hpp>

#include <iostream>

#include <fstream>

#include <sstream>

#include <mutex>

#include <tuple>

#include <net/if.h>
@@ -570,6 +571,8 @@ init_environment() { 


std::atomic<bool> init{false};

std::atomic<bool> preload_initializing{true};

static std::once_flag init_env_once;

static std::atomic<bool> env_initialized{false};



namespace gkfs::preload {
@@ -578,6 +581,59 @@ is_initializing() { 
    return preload_initializing.load(std::memory_order_acquire);

}



bool

is_environment_initialized() {

    return env_initialized.load(std::memory_order_acquire);

}



void

ensure_environment_initialized() {

    if(env_initialized.load(std::memory_order_acquire)) {

        return;

    }

    std::call_once(init_env_once, []() {

        LOG(INFO, "ensure_environment_initialized: triggering lazy GekkoFS client initialization...");

        preload_initializing.store(true, std::memory_order_release);

        if(CTX->protect_fds()) {

            CTX->protect_user_fds();

        }

        gkfs::preload::init_environment();

        if(CTX->protect_fds()) {

            CTX->unprotect_user_fds();

        }



        auto forwarding_map_file = gkfs::env::get_var(

                gkfs::env::FORWARDING_MAP_FILE, gkfs::config::forwarding_file_path);

        if(!forwarding_map_file.empty()) {

            init_forwarding_mapper();

        }



        if(!CTX->init_metrics()) {

            exit_error_msg(EXIT_FAILURE,

                           "Unable to initialize client metrics. Exiting...");

        }

        CTX->set_repair_executor([](const gkfs::rpc::repair_task& task) {

            if(task.kind == gkfs::rpc::repair_kind::data_chunk) {

                return gkfs::rpc::repair_data_copy(task);

            }

            if(task.kind == gkfs::rpc::repair_kind::metadata_create ||

               task.kind == gkfs::rpc::repair_kind::metadata_create_inline ||

               task.kind == gkfs::rpc::repair_kind::metadata_size ||

               task.kind == gkfs::rpc::repair_kind::metadata_inline) {

                return gkfs::rpc::repair_metadata_copy(task) == 0;

            }

            return false;

        });

        if(CTX->get_replicas() > 0 ||

           !gkfs::env::get_var(gkfs::env::REPAIR_JOURNAL_PATH).empty()) {

            CTX->start_repair_worker();

        }

        preload_initializing.store(false, std::memory_order_release);

        env_initialized.store(true, std::memory_order_release);

        LOG(INFO, "ensure_environment_initialized: GekkoFS client initialization complete.");

    });

}



} // namespace gkfs::preload



/**
@@ -661,6 +717,31 @@ init_preload() { 


    LOG(DEBUG, "Current working directory: '{}'", CTX->cwd());

    LOG(DEBUG, "Number of replicas : '{}'", CTX->get_replicas());



    // Setup mountdir early so path resolution can identify internal paths before Mercury init

    std::string mnt = gkfs::env::get_var(gkfs::env::MOUNTDIR, "");

    if(mnt.empty()) {

        mnt = gkfs::env::get_var("LIBGKFS_MOUNTDIR", "");

    }

    if(mnt.empty()) {

        std::string hostfile = gkfs::env::get_var(gkfs::env::HOSTS_FILE, gkfs::config::hostfile_path);

        if(!hostfile.empty()) {

            std::ifstream ifs(hostfile);

            std::string line;

            if(std::getline(ifs, line)) {

                std::istringstream iss(line);

                std::string h, u, p, m;

                if(iss >> h >> u >> p >> m && !m.empty()) {

                    mnt = m;

                }

            }

        }

    }

    if(!mnt.empty()) {

        CTX->mountdir(mnt);

        LOG(INFO, "Early mountdir registered: '{}'", CTX->mountdir());

    }



    if(std::getenv("LIBGKFS_KEEP_FI_PROVIDER_PATH") == nullptr) {

        unsetenv("FI_PROVIDER_PATH");

    }
@@ -670,19 +751,8 @@ init_preload() { 
    } else {

        setenv("HG_INIT_INFO", "request_post_init=64", 1);

    }

    gkfs::preload::init_environment();

    CTX->enable_interception();



    if(CTX->protect_fds())

        CTX->unprotect_user_fds();



    auto forwarding_map_file = gkfs::env::get_var(

            gkfs::env::FORWARDING_MAP_FILE, gkfs::config::forwarding_file_path);

    if(!forwarding_map_file.empty()) {

        init_forwarding_mapper();

    }



    // Special CONFIGURATION handling

    // Special CONFIGURATION handling (can be read early)

    gkfs::config::metadata::use_inline_data =

            gkfs::env::get_var(gkfs::env::USE_INLINE_DATA,

                               gkfs::config::metadata::use_inline_data
@@ -733,30 +803,11 @@ init_preload() { 
    LOG(INFO, "Local chunk bypass: {}", CTX->local_bypass() ? "ON" : "OFF");



#ifndef BYPASS_SYSCALL

    CTX->enable_interception();

    gkfs::preload::start_interception();

#endif

    errno = oerrno;



    if(!CTX->init_metrics()) {

        exit_error_msg(EXIT_FAILURE,

                       "Unable to initialize client metrics. Exiting...");

    }

    CTX->set_repair_executor([](const gkfs::rpc::repair_task& task) {

        if(task.kind == gkfs::rpc::repair_kind::data_chunk) {

            return gkfs::rpc::repair_data_copy(task);

        }

        if(task.kind == gkfs::rpc::repair_kind::metadata_create ||

           task.kind == gkfs::rpc::repair_kind::metadata_create_inline ||

           task.kind == gkfs::rpc::repair_kind::metadata_size ||

           task.kind == gkfs::rpc::repair_kind::metadata_inline) {

            return gkfs::rpc::repair_metadata_copy(task) == 0;

        }

        return false;

    });

    if(CTX->get_replicas() > 0 ||

       !gkfs::env::get_var(gkfs::env::REPAIR_JOURNAL_PATH).empty()) {

        CTX->start_repair_worker();

    }

    std::atexit(quick_exit_handler);

    preload_initializing.store(false, std::memory_order_release);

}
@@ -776,6 +827,10 @@ destroy_preload() { 
    if(init.exchange(false) == false)

        return; // Prevent double destruction



    if(!gkfs::preload::is_environment_initialized()) {

        return; // Fast exit: environment was never initialized

    }



    const auto& replica = gkfs::rpc::replica_metrics();

    const auto repair = CTX->repair_queue_stats();

    LOG(INFO,
@@ -888,14 +943,7 @@ gkfs_init() { 
        LOG(INFO, "Protecting user fds");

    }



    if(CTX->protect_fds()) {

        CTX->protect_user_fds();

    }

    gkfs::preload::init_environment();



    if(CTX->protect_fds())

        CTX->unprotect_user_fds();



    gkfs::preload::ensure_environment_initialized();

    return 0;

}