Commit 52ef5d1c authored by Ahmad Tarraf's avatar Ahmad Tarraf
Browse files

client: reload routing state after mutate without restarting 

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.
parent f170709b

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

+32 −1
Changes for include/client/preload_util.hpp: 32 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -121,7 +121,20 @@ 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 +143,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/gkfs_metadata.cpp

+2 −2
Changes for src/client/gkfs_metadata.cpp: 2 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -1196,7 +1196,7 @@ gkfs_opendir(const std::string& path) { 


        LOG(DEBUG,

            "{}() Sending async dirents for path '{}' to '{}' targets ...",

            __func__, path, CTX->hosts().size());

            __func__, path, CTX->hosts()->size());



        int cnt = 0;

        // Collect and process results
@@ -1289,7 +1289,7 @@ gkfs_opendir(const std::string& path) { 
                    in.start_key = "";

                    in.bulk_handle = exposed_buffers.back();

                    waiters.push_back(

                            get_dirents_rpc.on(CTX->hosts().at(targets[i]))

                            get_dirents_rpc.on(CTX->hosts()->at(targets[i]))

                                    .async(in));

                    waiter_buffer_ids.push_back(i);

                    waiter_servers.push_back(i);

src/client/malleability.cpp

+2 −2
Changes for src/client/malleability.cpp: 2 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -95,7 +95,7 @@ mutate_start(int old_server_conf, int new_server_conf, 
    // from its own state (which includes removing hosts parsed from markers).

    // We just need to validate that the marker file has the right number of

    // hosts.

    uint64_t detected_hosts = CTX->hosts().size();

    uint64_t detected_hosts = CTX->hosts()->size();

    bool marker_mode = false;

    if(hf && detected_hosts != static_cast<uint64_t>(old_server_conf)) {

        try {
@@ -112,7 +112,7 @@ mutate_start(int old_server_conf, int new_server_conf, 
        }

    }

    if(!marker_mode &&

       CTX->hosts().size() != static_cast<uint64_t>(old_server_conf)) {

       CTX->hosts()->size() != static_cast<uint64_t>(old_server_conf)) {

        auto err_str =

                "ERR: Old server configuration does not match the number of hosts in hostsfile";

        cerr << err_str << endl;