Commit 5b92785a authored by Ramon Nou's avatar Ramon Nou
Browse files

fix(daemon): track redistribution state atomically 

Use atomic load/store accessors for the redistribution running flag and set it before starting malleability redistribution threads. Reset the flag on ABT thread creation failure so daemon state remains consistent.
parent b3c7e3fb

src/daemon/classes/fs_data.cpp

+2 −2
Original line number Diff line number Diff line
@@ -384,12 +384,12 @@ FsData::maintenance_mode(bool maintenance_mode) { 


bool

FsData::redist_running() const {

    return redist_running_;

    return redist_running_.load();

}



void

FsData::redist_running(bool redist_running) {

    redist_running_ = redist_running;

    redist_running_.store(redist_running);

}



const std::shared_ptr<gkfs::malleable::MalleableManager>&

src/daemon/malleability/malleable_manager.cpp

+4 −0
Original line number Diff line number Diff line
@@ -909,10 +909,12 @@ MalleableManager::mutate_start(int old_server_conf, int new_server_conf, 
                "{}() {} active: old_hosts={}, new_hosts={}. Skipping eager data migration.",

                __func__, gkfs::env::EXPAND_ON_DEMAND, old_hosts_size_,

                hosts.size());

        GKFS_DATA->redist_running(true);

        auto abt_err =

                ABT_thread_create(RPC_DATA->io_pool(), expand_on_demand_abt,

                                  this, ABT_THREAD_ATTR_NULL, &redist_thread_);

        if(abt_err != ABT_SUCCESS) {

            GKFS_DATA->redist_running(false);

            throw runtime_error(fmt::format(

                    "MalleableManager::{}() Failed to create ABT thread with abt_err '{}'",

                    __func__, abt_err));
@@ -921,9 +923,11 @@ MalleableManager::mutate_start(int old_server_conf, int new_server_conf, 
    }



    // Use v2 pipeline for mutate

    GKFS_DATA->redist_running(true);

    auto abt_err = ABT_thread_create(RPC_DATA->io_pool(), expand_abt_v2, this,

                                     ABT_THREAD_ATTR_NULL, &redist_thread_);

    if(abt_err != ABT_SUCCESS) {

        GKFS_DATA->redist_running(false);

        auto err_str = fmt::format(

                "MalleableManager::{}() Failed to create ABT thread with abt_err '{}'",

                __func__, abt_err);