Commit 3e2a785b authored by Ramon Nou's avatar Ramon Nou
Browse files

Fix repair journal lifecycle and libc preload startup

parent f9451e03

include/client/preload.hpp

+3 −0
Changes for include/client/preload.hpp: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -47,6 +47,9 @@ 


#define CTX gkfs::preload::PreloadContext::getInstance()

namespace gkfs::preload {

bool

is_initializing();



void

init_environment();

void

include/client/preload_context.hpp

+3 −0
Changes for include/client/preload_context.hpp: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -514,6 +514,9 @@ public: 
    void

    stop_repair_worker();



    void

    release_repair_journal_lock();



    void

    repair_worker();

include/client/rpc/repair_journal_lock.hpp

+25 −3
Changes for include/client/rpc/repair_journal_lock.hpp: 25 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -27,6 +27,7 @@ 
#include <stdexcept>

#include <string>

#include <sys/types.h>

#include <thread>

#include <vector>



namespace gkfs::rpc {
@@ -116,6 +117,12 @@ public: 
            return;

        }



        std::ifstream existing_journal(journal_path);

        if(existing_journal.is_open() &&

           existing_journal.peek() == std::ifstream::traits_type::eof()) {

            return;

        }



        lock_path_ = journal_path + ".lock";

        const auto fd =

                ::open(lock_path_.c_str(), O_RDWR | O_CREAT | O_CLOEXEC, 0600);
@@ -131,7 +138,18 @@ public: 
        lock.l_whence = SEEK_SET;

        lock.l_start = 0;

        lock.l_len = 0;

        if(::fcntl(candidate.get(), F_OFD_SETLK, &lock) != 0) {

        auto lock_acquired = false;

        for(int attempt = 0; attempt < 20; ++attempt) {

            if(::fcntl(candidate.get(), F_OFD_SETLK, &lock) == 0) {

                lock_acquired = true;

                break;

            }

            if(errno != EWOULDBLOCK && errno != EAGAIN) {

                break;

            }

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

        }

        if(!lock_acquired) {

            const auto error = errno;

            if(error == EWOULDBLOCK || error == EAGAIN) {

                throw std::runtime_error(
@@ -142,10 +160,14 @@ public: 
                                     journal_path + "': " + ::strerror(error));

        }



        std::ifstream journal(journal_path);

        const bool journal_has_pending_records =

                !journal.is_open() ||

                journal.peek() != std::ifstream::traits_type::eof();

        if(const auto previous_topology =

                   read_repair_journal_topology_identity(lock_path_);

           previous_topology && !previous_topology->empty() &&

           topology_identity != "unknown" &&

           journal_has_pending_records && previous_topology &&

           !previous_topology->empty() && topology_identity != "unknown" &&

           *previous_topology != topology_identity) {

            throw std::runtime_error(

                    "Replica repair journal topology mismatch: expected '" +

src/client/gkfs_libc.cpp

+88 −85

File changed.

Preview size limit exceeded, changes collapsed.

src/client/preload.cpp

+14 −18
Changes for src/client/preload.cpp: 14 added lines, 18 removed lines.
Original line number Diff line number Diff line
@@ -534,6 +534,16 @@ init_environment() { 




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

std::atomic<bool> preload_initializing{true};



namespace gkfs::preload {



bool

is_initializing() {

    return preload_initializing.load(std::memory_order_acquire);

}



} // namespace gkfs::preload



/**

 * Called initially ONCE when preload library is used with the LD_PRELOAD
@@ -576,6 +586,7 @@ init_preload() { 
    // The original errno value will be restored after initialization to not

    // leak internal error codes

    auto oerrno = errno;

    preload_initializing.store(true, std::memory_order_release);

    init = true;

    pthread_atfork(&at_fork, &at_parent, &at_child);
@@ -687,6 +698,7 @@ init_preload() { 
    });

    CTX->start_repair_worker();

    std::atexit(quick_exit_handler);

    preload_initializing.store(false, std::memory_order_release);

}
@@ -731,6 +743,7 @@ destroy_preload() { 
        CTX->stop_async_write_thread();

    }

    CTX->stop_repair_worker();

    CTX->release_repair_journal_lock();

    auto forwarding_map_file = gkfs::env::get_var(

            gkfs::env::FORWARDING_MAP_FILE, gkfs::config::forwarding_file_path);

    if(!forwarding_map_file.empty()) {
@@ -802,24 +815,7 @@ gkfs_init() { 


extern "C" int

gkfs_end() {

    CTX->clear_hosts();

    LOG(DEBUG, "Peer information deleted");



    if(CTX->use_proxy()) {

        CTX->clear_proxy_host();

        LOG(DEBUG, "Shutting down IPC subsystem");

        CTX->ipc_engine(nullptr);

    }



    LOG(DEBUG, "Shutting down RPC subsystem");

    CTX->rpc_engine(nullptr);

    LOG(DEBUG, "RPC subsystem shut down");



    // ld_margo_rpc_id = MARGO_INSTANCE_NULL;

    // ld_margo_ipc_id = MARGO_INSTANCE_NULL;



    LOG(INFO, "All subsystems shut down. Client shutdown complete.");



    destroy_preload();

    return 0;

}