Commit 445f014f authored by Ramon Nou's avatar Ramon Nou
Browse files

refactor: own daemon shutdown workers

parent e369175d

WIP.md

+7 −5
Original line number Diff line number Diff line
@@ -424,12 +424,14 @@ subsystem boundaries, preserve causes and operation context, centralize POSIX 
and RPC conversion, and mark potentially failing functions. Logs and returned

errors must agree, especially for retryable transport failures.



### 23. Modernize resource ownership and shutdown

### 23. Modernize resource ownership and shutdown — in progress



Add RAII wrappers for ABT threads, RPC engines, descriptors, buffers, temporary

files, and backend handles where practical. Define shutdown ordering for preload

destructors, daemon signals, proxy shutdown, migration workers, and logging.

Avoid detached work that outlives its owner. Run leak, sanitizer, and repeated

Added explicit ownership cleanup for the daemon malleability ABT worker and

ordered its destruction before Argobots execution streams. Replaced the

detached delayed daemon-shutdown thread with a joinable daemon-owned thread.



Remaining work: broader RAII wrappers for RPC engines, descriptors, buffers,

temporary files, and backend handles; sanitizer/leak coverage; and repeated

initialization/shutdown tests.



### 24. Simplify and standardize CMake configuration

include/daemon/daemon.hpp

+7 −0
Original line number Diff line number Diff line
@@ -44,6 +44,7 @@ 


// std libs

#include <string>

#include <chrono>

#include <spdlog/spdlog.h>



#include <config.hpp>
@@ -90,6 +91,12 @@ namespace gkfs::daemon { 
void

request_shutdown();



void

schedule_shutdown(std::chrono::milliseconds delay);



void

join_scheduled_shutdown();



} // namespace gkfs::daemon



#endif // GKFS_DAEMON_DAEMON_HPP

include/daemon/malleability/malleable_manager.hpp

+9 −1
Original line number Diff line number Diff line
@@ -46,7 +46,10 @@ namespace gkfs::malleable { 


class MalleableManager {

private:

    ABT_thread redist_thread_;

    ABT_thread redist_thread_{ABT_THREAD_NULL};



    void

    join_redist_thread();



    // Tracks old hosts_size before expansion/shrink

    unsigned int old_hosts_size_{0};
@@ -80,6 +83,11 @@ private: 
    expand_on_demand_abt(void* _arg);



public:

    ~MalleableManager();



    void

    wait_for_worker();



    void

    mutate_start(int old_server_conf, int new_server_conf,

                 const std::string& new_hosts_file);

src/daemon/daemon.cpp

+25 −0
Original line number Diff line number Diff line
@@ -107,6 +107,7 @@ struct rpc_err_out_t; 


static condition_variable shutdown_please; // handler for shutdown signaling

static mutex mtx; // mutex to wait on shutdown conditional variable

static thread scheduled_shutdown_thread;

static bool keep_rootdir = true;



namespace {
@@ -892,11 +893,17 @@ void 
destroy_enviroment() {

    std::error_code ecode;

    remove_daemon_readiness();

    gkfs::daemon::join_scheduled_shutdown();

#ifdef GKFS_ENABLE_CLIENT_METRICS

    stop_metrics_relay();

#endif

    GKFS_DATA->spdlogger()->debug("{}() Freeing I/O executions streams",

                                  __func__);

    if(GKFS_DATA->malleable_manager()) {

        auto malleable_manager = GKFS_DATA->malleable_manager();

        malleable_manager->wait_for_worker();

        GKFS_DATA->malleable_manager(nullptr);

    }

    for(unsigned int i = 0; i < RPC_DATA->io_streams().size(); i++) {

        ABT_xstream_join(RPC_DATA->io_streams().at(i));

        ABT_xstream_free(&RPC_DATA->io_streams().at(i));
@@ -949,6 +956,24 @@ request_shutdown() { 
    shutdown_please.notify_all();

}



void

schedule_shutdown(const std::chrono::milliseconds delay) {

    if(scheduled_shutdown_thread.joinable()) {

        scheduled_shutdown_thread.join();

    }

    scheduled_shutdown_thread = std::thread([delay]() {

        std::this_thread::sleep_for(delay);

        request_shutdown();

    });

}



void

join_scheduled_shutdown() {

    if(scheduled_shutdown_thread.joinable()) {

        scheduled_shutdown_thread.join();

    }

}



} // namespace gkfs::daemon



/**

src/daemon/handler/srv_malleability.cpp

+1 −5
Original line number Diff line number Diff line
@@ -44,7 +44,6 @@ 
#include <daemon/backend/metadata/db.hpp>



#include <chrono>

#include <thread>



extern "C" {

#include <unistd.h>
@@ -288,10 +287,7 @@ rpc_srv_mutate_shutdown(const tl::request& req, 
    gkfs::utils::safe_respond(req, out, __func__);



    if(out.err == 0) {

        std::thread([]() {

            std::this_thread::sleep_for(std::chrono::milliseconds(100));

            gkfs::daemon::request_shutdown();

        }).detach();

        gkfs::daemon::schedule_shutdown(std::chrono::milliseconds(100));

    }

}