client: reload routing state after mutate without restarting

Ramon Nourequested to merge
client-mutate-reload into master

A completed gkfs_malleability mutate updates the daemons via forward_mutate_reload() but never an already-connected client, which keeps routing RPCs to removed daemons forever (NA_TIMEOUT retry storm, confirmed via live pstack against DMR malleable-checkpoint workloads on MN5).

Adds an in-process reload: PreloadContext now holds hosts/distributor/ local_host_id as one mutex-guarded RoutingSnapshot swapped atomically via commit_routing(), and a dedicated watcher thread (sigwait on LIBGKFS_RELOAD_SIGNAL, default SIGUSR1, blocked process-wide so it can only be consumed there) calls reload_client_topology() to reparse the hosts file and rebuild routing state live -- mirroring MalleableManager::reload_hosts_file() on the daemon side, without tearing down the Margo/Mercury engine or restarting the process.

Merge request reports

include/client/env.hpp

+5 −0
Changes for include/client/env.hpp: 5 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -62,6 +62,11 @@ static constexpr auto LOG_SAMPLE_RATE = ADD_PREFIX("LOG_SAMPLE_RATE"); 
static constexpr auto CWD = ADD_PREFIX("CWD");

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 +

// distributor) after a `gkfs_malleability mutate` completes, without

// restarting the process. Configurable in case the host application already

// uses the default (SIGUSR1) for its own purposes.

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

#ifdef GKFS_ENABLE_CLIENT_METRICS

static constexpr auto METRICS_FLUSH_INTERVAL =

        ADD_PREFIX("METRICS_FLUSH_INTERVAL");

include/client/preload_context.hpp

+62 −11
Changes for include/client/preload_context.hpp: 62 added lines, 11 removed lines.
Original line number Diff line number Diff line
@@ -52,6 +52,7 @@ 
#include <vector>

#include <string>

#include <mutex>

#include <atomic>

#include <unordered_set>

#include <config.hpp>

#include <client/rpc/repair_queue.hpp>
@@ -116,6 +117,20 @@ struct FsConfig { 


enum class RelativizeStatus { internal, external, fd_unknown, fd_not_a_dir };



/**

 * Immutable snapshot of a client's daemon routing state: which daemons it

 * talks to, how it maps a path to one of them, and which index in `hosts` is

 * this process itself. Rebuilt whole and swapped atomically by

 * commit_routing() -- readers on any thread get a consistent view (never a

 * distributor built against a hosts list from before/after it) without

 * blocking on RPC I/O for the swap itself.

 */

struct RoutingSnapshot {

    std::vector<thallium::endpoint> hosts;

    std::shared_ptr<gkfs::rpc::Distributor> distributor;

    uint64_t local_host_id{0};

};



/**

 * Singleton class of the client context with all relevant global data

 */
@@ -129,7 +144,6 @@ private: 
    PreloadContext();



    std::shared_ptr<gkfs::filemap::OpenFileMap> ofm_;

    std::shared_ptr<gkfs::rpc::Distributor> distributor_;

    std::shared_ptr<FsConfig> fs_conf_;

    std::shared_ptr<gkfs::cache::dir::DentryCache> dentry_cache_;

    bool use_dentry_cache_{false};
@@ -141,9 +155,15 @@ private: 
    std::vector<std::string> mountdir_components_;

    std::string mountdir_;



    std::vector<thallium::endpoint> hosts_;

    // Daemon routing state. Never touch these three fields directly -- go

    // through commit_routing()/the read accessors, which are the only things

    // that keep them consistent with each other under concurrent RPC traffic

    // (see RoutingSnapshot).

    std::shared_ptr<const RoutingSnapshot> routing_{

            std::make_shared<const RoutingSnapshot>()};

    mutable std::mutex routing_mutex_;



    thallium::endpoint proxy_host_;

    uint64_t local_host_id_;

    uint64_t fwd_host_id_;

    std::string rpc_protocol_;

    bool auto_sm_{false};
@@ -191,6 +211,14 @@ private: 
    std::thread async_write_thread_;

    bool async_write_stop_{false};



    // Watches for LIBGKFS_RELOAD_SIGNAL (default SIGUSR1, blocked

    // process-wide so it only ever arrives here via sigwait()) and reloads

    // routing_ from the current hosts file when it fires -- see

    // gkfs::utils::reload_client_topology().

    std::thread reload_watcher_thread_;

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

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



    gkfs::rpc::repair_queue repair_queue_;

    mutable std::mutex repair_queue_mutex_;

    std::condition_variable repair_queue_cv_;
@@ -248,11 +276,21 @@ public: 
    const std::string&

    cwd() const;



    const std::vector<thallium::endpoint>&

    std::shared_ptr<const std::vector<thallium::endpoint>>

    hosts() const;



    /**

     * Replaces the daemon routing state (hosts, distributor, and this

     * process's index into hosts) as one atomic unit. This is the ONLY way

     * routing_ is mutated -- init_environment() and

     * gkfs::utils::reload_client_topology() both funnel through it, so a

     * reader can never observe a distributor built against a hosts list

     * from a different generation.

     */

    void

    hosts(const std::vector<thallium::endpoint>& addrs);

    commit_routing(std::vector<thallium::endpoint> hosts,

                   std::shared_ptr<gkfs::rpc::Distributor> distributor,

                   uint64_t local_host_id);



    void

    clear_hosts();
@@ -260,9 +298,6 @@ public: 
    uint64_t

    local_host_id() const;



    void

    local_host_id(uint64_t id);



    uint64_t

    fwd_host_id() const;
@@ -314,9 +349,6 @@ public: 
    const std::shared_ptr<gkfs::filemap::OpenFileMap>&

    file_map() const;



    void

    distributor(std::shared_ptr<gkfs::rpc::Distributor> distributor);



    std::shared_ptr<gkfs::rpc::Distributor>

    distributor() const;
@@ -514,6 +546,25 @@ public: 
    void

    async_write_worker();



    /**

     * Blocks LIBGKFS_RELOAD_SIGNAL (default SIGUSR1) on the calling thread --

     * call once, early, before any other thread (Margo/Argobots/FUSE) is

     * spawned, so every thread this process ever creates inherits the

     * blocked mask and the signal can only ever be consumed by

     * reload_watcher_worker()'s sigwait().

     */

    void

    block_reload_signal();



    void

    start_reload_watcher_thread();



    void

    stop_reload_watcher_thread();



    void

    reload_watcher_worker();



    void

    enqueue_async_write(const std::string& path, const void* buf,

                        off64_t offset, size_t count, int8_t num_copies);

include/client/preload_util.hpp

+33 −1
Changes for include/client/preload_util.hpp: 33 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -121,7 +121,21 @@ load_forwarding_map(); 
std::vector<std::pair<std::string, std::string>>

read_hosts_file();



void

/**

 * @brief Resolves Mercury/Thallium addresses for a set of daemons and finds

 * this process's own position among them.

 * @param hosts vector<pair<hostname, Mercury URI address>>

 * @return {resolved endpoints (same order/index as `hosts`), this host's

 * index into that vector}

 * @throws std::runtime_error through lookup_endpoint()

 *

 * Pure computation -- does not touch PreloadContext. Callers

 * (init_environment() and reload_client_topology()) are responsible for

 * committing the result via PreloadContext::commit_routing(), together with a

 * distributor built from the same (hosts, local_host_id) pair, so the three

 * never drift out of sync.

 */

std::pair<std::vector<thallium::endpoint>, uint64_t>

connect_to_hosts(const std::vector<std::pair<std::string, std::string>>& hosts);



void
@@ -130,6 +144,24 @@ check_for_proxy(); 
void

lookup_proxy_addr();



/**

 * @brief Reparses the current hosts file and rebuilds this client's routing

 * state (hosts + distributor) in place, then commits it atomically via

 * PreloadContext::commit_routing() -- without restarting the process or

 * tearing down the Margo/Mercury engine.

 *

 * Mirrors MalleableManager::reload_hosts_file() on the daemon side. Called

 * from PreloadContext::reload_watcher_worker() after LIBGKFS_RELOAD_SIGNAL

 * fires (default SIGUSR1), which our own launcher scripts send to every live

 * client right after `gkfs_malleability mutate finalize` completes.

 *

 * Refuses to run (logs and returns, leaving the current routing state

 * untouched) if the hosts file still carries '+'/'-' markers -- i.e. if a

 * mutate is still in progress rather than finalized.

 */

void

reload_client_topology();



} // namespace gkfs::utils

src/client/rpc/forward_data.cpp

+14 −14
Changes for src/client/rpc/forward_data.cpp: 14 added lines, 14 removed lines.
Original line number Diff line number Diff line
@@ -106,7 +106,7 @@ forward_data_helper(std::string rpc_tag, 
                    in.trace_id, rpc_tag, target);

            }

            waiters.push_back(forward_async_with_timeout(

                    rpc, CTX->hosts().at(target), in));

                    rpc, CTX->hosts()->at(target), in));

        } catch(const std::exception& ex) {

            LOG(ERROR, "Failed to send RPC to host {}: {}", target, ex.what());

        }
@@ -280,7 +280,7 @@ forward_write(const string& path, const void* buf, const off64_t offset, 
        in.path = path;

        in.offset = block_overrun(offset, gkfs::config::rpc::chunksize);

        in.host_id = target;

        in.host_size = CTX->hosts().size();

        in.host_size = CTX->hosts()->size();

        in.wbitset = gkfs::rpc::compress_bitset(chnk_bitset);

        in.chunk_n = target_chnks[target].size();

        in.chunk_start = chnk_start;
@@ -293,11 +293,11 @@ forward_write(const string& path, const void* buf, const off64_t offset, 
                       num_copies > 0 ? num_copies : CTX->get_replicas(),

                       CTX->rpc_timeout())) {

                waiters.push_back(

                        write_rpc.on(CTX->hosts().at(target))

                        write_rpc.on(CTX->hosts()->at(target))

                                .timed_async(std::chrono::seconds(5), in));

            } else {

                waiters.push_back(forward_async_with_timeout(

                        write_rpc, CTX->hosts().at(target), in));

                        write_rpc, CTX->hosts()->at(target), in));

            }

            waiter_targets.push_back(target);

        } catch(const std::exception& ex) {
@@ -519,7 +519,7 @@ forward_read(const string& path, void* buf, const off64_t offset, 
        in.path = path;

        in.offset = block_overrun(offset, gkfs::config::rpc::chunksize);

        in.host_id = target;

        in.host_size = CTX->hosts().size();

        in.host_size = CTX->hosts()->size();

        in.wbitset = gkfs::rpc::compress_bitset(chnk_bitset);

        in.chunk_n = target_chnks[target].size();

        in.chunk_start = chnk_start;
@@ -531,11 +531,11 @@ forward_read(const string& path, void* buf, const off64_t offset, 
            if(gkfs::rpc::use_replica_rpc_timeout(num_copies,

                                                  CTX->rpc_timeout())) {

                waiters.push_back(

                        read_rpc.on(CTX->hosts().at(target))

                        read_rpc.on(CTX->hosts()->at(target))

                                .timed_async(std::chrono::seconds(5), in));

            } else {

                waiters.push_back(forward_async_with_timeout(

                        read_rpc, CTX->hosts().at(target), in));

                        read_rpc, CTX->hosts()->at(target), in));

            }

            waiter_targets.push_back(target);

        } catch(const std::exception& ex) {
@@ -637,7 +637,7 @@ repair_data_copy(const gkfs::rpc::repair_task& task) { 
        read_in.offset = 0;

        read_in.host_id = source->host;

        read_in.copy = task.source_copy;

        read_in.host_size = CTX->hosts().size();

        read_in.host_size = CTX->hosts()->size();

        read_in.wbitset = gkfs::rpc::compress_bitset(std::vector<uint8_t>{1});

        read_in.chunk_n = 1;

        read_in.chunk_start = task.chunk_id;
@@ -646,7 +646,7 @@ repair_data_copy(const gkfs::rpc::repair_task& task) { 
        read_in.bulk_handle = read_bulk;



        const auto read_out =

                forward_with_timeout(read_rpc, CTX->hosts().at(source->host),

                forward_with_timeout(read_rpc, CTX->hosts()->at(source->host),

                                     read_in)

                        .template as<gkfs::rpc::rpc_data_out_t>();

        if(read_out.err != 0 || read_out.io_size == 0 ||
@@ -666,7 +666,7 @@ repair_data_copy(const gkfs::rpc::repair_task& task) { 
        write_in.path = task.path;

        write_in.offset = 0;

        write_in.host_id = target->host;

        write_in.host_size = CTX->hosts().size();

        write_in.host_size = CTX->hosts()->size();

        write_in.wbitset = gkfs::rpc::compress_bitset(std::vector<uint8_t>{1});

        write_in.chunk_n = 1;

        write_in.chunk_start = task.chunk_id;
@@ -675,7 +675,7 @@ repair_data_copy(const gkfs::rpc::repair_task& task) { 
        write_in.bulk_handle = write_bulk;



        const auto write_out =

                forward_with_timeout(write_rpc, CTX->hosts().at(target->host),

                forward_with_timeout(write_rpc, CTX->hosts()->at(target->host),

                                     write_in)

                        .template as<gkfs::rpc::rpc_data_out_t>();

        if(write_out.err == 0 && write_out.io_size == read_out.io_size) {
@@ -759,8 +759,8 @@ forward_get_chunk_stat() { 


    auto unavailable = CTX->unavailable_hosts();

    std::vector<uint64_t> hosts;

    hosts.reserve(CTX->hosts().size());

    for(std::size_t i = 0; i < CTX->hosts().size(); ++i) {

    hosts.reserve(CTX->hosts()->size());

    for(std::size_t i = 0; i < CTX->hosts()->size(); ++i) {

        if(unavailable.count(i) == 0) {

            hosts.push_back(i);

        }
@@ -776,7 +776,7 @@ forward_get_chunk_stat() { 
        in.dummy = 0;

        try {

            const auto out =

                    forward_with_timeout(stat_rpc, CTX->hosts().at(host), in)

                    forward_with_timeout(stat_rpc, CTX->hosts()->at(host), in)

                            .template as<gkfs::rpc::rpc_chunk_stat_out_t>();

            if(out.err != 0) {

                CTX->mark_host_failure(host);

src/client/rpc/forward_malleability.cpp

+1 −1
Changes for src/client/rpc/forward_malleability.cpp: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -66,7 +66,7 @@ mutate_endpoints_from_loaded_hosts() { 
    for(auto target : targets) {

        try {

            endpoints.emplace_back(std::to_string(target),

                                   CTX->hosts().at(target));

                                   CTX->hosts()->at(target));

        } catch(const std::exception& ex) {

            LOG(ERROR, "Failed to get mutate endpoint for host {}: {}", target,

                ex.what());