Commit b1409d11 authored by Ramon Nou's avatar Ramon Nou
Browse files

refactor: use RAII for migration file descriptors

parent 445f014f

WIP.md

+2 −0
Original line number Diff line number Diff line
@@ -429,6 +429,8 @@ errors must agree, especially for retryable transport failures. 
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.

Added a common move-only `unique_fd` owner and migrated migration checkpoint

and atomic chunk temporary-file writes to it.



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

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

include/common/unique_fd.hpp

0 → 100644
+101 −0
Original line number Diff line number Diff line 
/*

  Copyright 2018-2025, Barcelona Supercomputing Center (BSC), Spain

  Copyright 2015-2025, Johannes Gutenberg Universitaet Mainz, Germany



  This file is part of GekkoFS.



  GekkoFS is free software: you can redistribute it and/or modify

  it under the terms of the GNU General Public License as published by

  the Free Software Foundation, either version 3 of the License, or

  (at your option) any later version.



  GekkoFS is distributed in the hope that it will be useful,

  but WITHOUT ANY WARRANTY; without even the implied warranty of

  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

  GNU General Public License for more details.



  SPDX-License-Identifier: GPL-3.0-or-later

*/



#ifndef GEKKOFS_COMMON_UNIQUE_FD_HPP

#define GEKKOFS_COMMON_UNIQUE_FD_HPP



#include <unistd.h>



#include <utility>



namespace gkfs::utils {



/**

 * @brief Move-only RAII owner for a POSIX file descriptor.

 */

class unique_fd {

public:

    unique_fd() noexcept = default;



    explicit unique_fd(const int fd) noexcept : fd_{fd} {}



    unique_fd(const unique_fd&) = delete;

    unique_fd&

    operator=(const unique_fd&) = delete;



    unique_fd(unique_fd&& other) noexcept : fd_{other.release()} {}



    unique_fd&

    operator=(unique_fd&& other) noexcept {

        if(this != &other) {

            reset(other.release());

        }

        return *this;

    }



    ~unique_fd() {

        reset();

    }



    [[nodiscard]] bool

    valid() const noexcept {

        return fd_ >= 0;

    }



    [[nodiscard]] int

    get() const noexcept {

        return fd_;

    }



    [[nodiscard]] explicit

    operator bool() const noexcept {

        return valid();

    }



    int

    release() noexcept {

        const auto fd = fd_;

        fd_ = -1;

        return fd;

    }



    void

    reset(const int fd = -1) noexcept {

        if(fd_ >= 0) {

            ::close(fd_);

        }

        fd_ = fd;

    }



    int

    close() noexcept {

        if(fd_ < 0) {

            return 0;

        }

        const auto fd = release();

        return ::close(fd);

    }



private:

    int fd_{-1};

};



} // namespace gkfs::utils



#endif // GEKKOFS_COMMON_UNIQUE_FD_HPP
 
 No newline at end of file

src/daemon/backend/data/chunk_storage.cpp

+7 −8
Original line number Diff line number Diff line
@@ -44,6 +44,7 @@ 
#include <daemon/backend/data/chunk_storage.hpp>

#include <daemon/backend/data/file_handle.hpp>

#include <common/path_util.hpp>

#include <common/unique_fd.hpp>



#include <cerrno>

#include <algorithm>
@@ -254,9 +255,9 @@ ChunkStorage::write_chunk_atomic(const string& file_path, 
    const auto temporary_path = fmt::format(

            "{}.migration.tmp.{}.{}", chunk_path, getpid(),

            migration_tmp_seq.fetch_add(1, std::memory_order_relaxed));

    const auto fd =

            open(temporary_path.c_str(), O_WRONLY | O_CREAT | O_EXCL, 0640);

    if(fd < 0) {

    gkfs::utils::unique_fd fd(

            open(temporary_path.c_str(), O_WRONLY | O_CREAT | O_EXCL, 0640));

    if(!fd) {

        throw ChunkStorageException(

                errno,

                fmt::format(
@@ -266,13 +267,12 @@ ChunkStorage::write_chunk_atomic(const string& file_path, 


    size_t written = 0;

    while(written < size) {

        const auto ret = write(fd, buf + written, size - written);

        const auto ret = write(fd.get(), buf + written, size - written);

        if(ret < 0 && errno == EINTR) {

            continue;

        }

        if(ret <= 0) {

            const auto error = errno;

            close(fd);

            unlink(temporary_path.c_str());

            throw ChunkStorageException(

                    error,
@@ -282,15 +282,14 @@ ChunkStorage::write_chunk_atomic(const string& file_path, 
        }

        written += static_cast<size_t>(ret);

    }

    if(fsync(fd) != 0) {

    if(fsync(fd.get()) != 0) {

        const auto error = errno;

        close(fd);

        unlink(temporary_path.c_str());

        throw ChunkStorageException(

                error, fmt::format("Failed to fsync migration chunk '{}': {}",

                                   temporary_path, strerror(error)));

    }

    if(close(fd) != 0) {

    if(fd.close() != 0) {

        const auto error = errno;

        unlink(temporary_path.c_str());

        throw ChunkStorageException(

src/daemon/malleability/malleable_manager.cpp

+6 −6
Original line number Diff line number Diff line
@@ -46,6 +46,7 @@ 
#include <common/rpc/random_slicing_distributor.hpp>

#include <common/rpc/cutshift_sorted.hpp>

#include <common/hostfile_management.hpp>

#include <common/unique_fd.hpp>

#include <common/env.hpp>



#include <filesystem>
@@ -101,23 +102,22 @@ write_migration_checkpoint(const std::string& state, size_t jobs_total, 
                state, getpid(), GKFS_DATA->hosts_file(), jobs_total,

                jobs_completed, jobs_succeeded, jobs_failed, bytes_transferred,

                error);

        const auto fd =

                open(temporary.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0640);

        if(fd < 0) {

        gkfs::utils::unique_fd fd(

                open(temporary.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0640));

        if(!fd) {

            throw std::runtime_error(

                    fmt::format("Failed to open migration checkpoint '{}': {}",

                                temporary, strerror(errno)));

        }

        size_t written = 0;

        while(written < content.size()) {

            const auto ret = ::write(fd, content.data() + written,

            const auto ret = ::write(fd.get(), content.data() + written,

                                     content.size() - written);

            if(ret < 0 && errno == EINTR) {

                continue;

            }

            if(ret <= 0) {

                const auto error_code = errno;

                ::close(fd);

                ::unlink(temporary.c_str());

                throw std::runtime_error(fmt::format(

                        "Failed to write migration checkpoint '{}': {}",
@@ -125,7 +125,7 @@ write_migration_checkpoint(const std::string& state, size_t jobs_total, 
            }

            written += static_cast<size_t>(ret);

        }

        if(::fsync(fd) != 0 || ::close(fd) != 0) {

        if(::fsync(fd.get()) != 0 || fd.close() != 0) {

            const auto error_code = errno;

            ::unlink(temporary.c_str());

            throw std::runtime_error(fmt::format(

tests/unit/CMakeLists.txt

+1 −0
Original line number Diff line number Diff line
@@ -64,6 +64,7 @@ target_sources(unit_tests 
    ${CMAKE_CURRENT_LIST_DIR}/test_random_slicing_pipeline.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_env_util.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_stats.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_unique_fd.cpp

    ${CMAKE_SOURCE_DIR}/src/common/hostfile_management.cpp)



if (GKFS_TESTS_GUIDED_DISTRIBUTION)