Commit 4983269a authored by Ramon Nou's avatar Ramon Nou
Browse files

add atomic daemon readiness state

parent 8b399a02

src/daemon/daemon.cpp

+70 −1
Original line number Diff line number Diff line
@@ -107,6 +107,18 @@ static condition_variable shutdown_please; // handler for shutdown signaling 
static mutex mtx; // mutex to wait on shutdown conditional variable

static bool keep_rootdir = true;



namespace {



constexpr auto daemon_readiness_filename = "gkfs.ready";



void

publish_daemon_readiness();



void

remove_daemon_readiness();



} // namespace



#ifdef GKFS_ENABLE_CLIENT_METRICS

namespace {

// test-only opt-in feature (GKFS_ENABLE_METRICS + GKFS_METRICS_AGGREGATOR).
@@ -715,7 +727,11 @@ init_environment() { 
    start_metrics_relay();

#endif



    GKFS_DATA->spdlogger()->info("Startup successful. Daemon is ready.");

    publish_daemon_readiness();

    GKFS_DATA->spdlogger()->info(

            "Startup successful. Daemon is ready (readiness_file='{}').",

            (fs::path(GKFS_DATA->rootdir()) / daemon_readiness_filename)

                    .native());

}



#ifdef GKFS_ENABLE_AGIOS
@@ -738,6 +754,58 @@ agios_initialize() { 
}

#endif



namespace {



void

publish_daemon_readiness() {

    const fs::path readiness_path =

            fs::path(GKFS_DATA->rootdir()) / daemon_readiness_filename;

    const fs::path temporary_path = readiness_path.string() + ".tmp";

    std::ofstream readiness(temporary_path, std::ios::trunc);

    if(!readiness.is_open()) {

        throw std::runtime_error(

                fmt::format("Failed to create daemon readiness file '{}'",

                            temporary_path.native()));

    }

    readiness << "state=ready\n"

              << "pid=" << ::getpid() << '\n'

              << "host_id=" << RPC_DATA->local_host_id() << '\n'

              << "rpc_address=" << GKFS_DATA->bind_addr() << '\n'

              << "hostfile=" << GKFS_DATA->hosts_file() << '\n';

    readiness.flush();

    if(!readiness) {

        throw std::runtime_error(

                fmt::format("Failed to write daemon readiness file '{}'",

                            temporary_path.native()));

    }

    readiness.close();

    std::error_code error;

    fs::rename(temporary_path, readiness_path, error);

    if(error) {

        fs::remove(temporary_path);

        throw std::runtime_error(

                fmt::format("Failed to publish daemon readiness file '{}': {}",

                            readiness_path.native(), error.message()));

    }

}



void

remove_daemon_readiness() {

    if(GKFS_DATA->rootdir().empty()) {

        return;

    }

    std::error_code error;

    fs::remove(fs::path(GKFS_DATA->rootdir()) / daemon_readiness_filename,

               error);

    if(error) {

        GKFS_DATA->spdlogger()->debug(

                "{}() Failed to remove readiness file: {}", __func__,

                error.message());

    }

}



} // namespace



/**

 * @brief Destroys the daemon environment and gracefully shuts down all

 * subroutines.
@@ -749,6 +817,7 @@ agios_initialize() { 
void

destroy_enviroment() {

    std::error_code ecode;

    remove_daemon_readiness();

#ifdef GKFS_ENABLE_CLIENT_METRICS

    stop_metrics_relay();

#endif

tests/integration/harness/gkfs.py

+22 −0
Original line number Diff line number Diff line
@@ -395,12 +395,23 @@ class Daemon: 
        """



        gkfs_daemon_active_log_pattern = r'Startup successful. Daemon is ready.'

        readiness_path = self.rootdir / 'gkfs.ready'



        init_time = perf_counter()



        while perf_counter() - init_time < timeout:

            if self._proc.poll() is not None:

                 raise RuntimeError(f"process {self._proc.pid} exited with {self._proc.returncode}")

            try:

                readiness = readiness_path.read_text()

                fields = dict(

                        line.split('=', 1)

                        for line in readiness.splitlines()

                        if '=' in line)

                if fields.get('state') == 'ready' and int(fields.get('pid', -1)) == pid:

                    return

            except (FileNotFoundError, ValueError):

                pass

            try:

                log_path = self.logdir / gkfs_daemon_log_file
@@ -1520,11 +1531,22 @@ class FwdDaemon: 
        """



        gkfs_daemon_active_log_pattern = r'Startup successful. Daemon is ready.'

        readiness_path = self.rootdir / 'gkfs.ready'



        init_time = perf_counter()



        while perf_counter() - init_time < timeout:

            try:

                readiness = readiness_path.read_text()

                fields = dict(

                        line.split('=', 1)

                        for line in readiness.splitlines()

                        if '=' in line)

                if fields.get('state') == 'ready' and int(fields.get('pid', -1)) == pid:

                    return

            except (FileNotFoundError, ValueError):

                pass

            try:

#                logger.debug(f"checking log file")

                with open(self.logdir / gkfwd_daemon_log_file) as log:

                    for line in islice(log, max_lines):