Commit 0e997ade authored by Ramon Nou's avatar Ramon Nou
Browse files

Weight Random Slicing by storage performance and available space

parent 8b0490b4

README.md

+5 −0
Changes for README.md: 5 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -622,6 +622,11 @@ export GKFS_DISTRIBUTION_STRATEGY=simple_hash 
# export GKFS_DISTRIBUTION_STRATEGY=random_slicing

# export GKFS_RANDOM_SLICING_CUTSHIFT=ON



# optional Random Slicing startup weights (set on every daemon before startup):

# export GKFS_RANDOM_SLICING_WEIGHT_PERF=ON   # calibrated write throughput

# export GKFS_RANDOM_SLICING_WEIGHT_SPACE=ON  # available bytes on the daemon root filesystem

# Enable either option alone or both together; both OFF gives equal weights.



# recommended on multi-node systems to avoid clients choosing loopback

export LIBGKFS_OFI_INTERFACE=ib0

```

include/common/env.hpp

+4 −2
Changes for include/common/env.hpp: 4 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -58,8 +58,10 @@ static constexpr auto DISTRIBUTION_STRATEGY = 
        ADD_PREFIX("DISTRIBUTION_STRATEGY");

static constexpr auto RANDOM_SLICING_CUTSHIFT =

        ADD_PREFIX("RANDOM_SLICING_CUTSHIFT");

static constexpr auto RANDOM_SLICING_CALIBRATION =

        ADD_PREFIX("RANDOM_SLICING_CALIBRATION");

static constexpr auto RANDOM_SLICING_WEIGHT_PERF =

        ADD_PREFIX("RANDOM_SLICING_WEIGHT_PERF");

static constexpr auto RANDOM_SLICING_WEIGHT_SPACE =

        ADD_PREFIX("RANDOM_SLICING_WEIGHT_SPACE");

static constexpr auto RANDOM_SLICING_CALIBRATION_SECONDS =

        ADD_PREFIX("RANDOM_SLICING_CALIBRATION_SECONDS");

static constexpr auto RANDOM_SLICING_CALIBRATION_MIN_MBPS =

include/common/rpc/random_slicing_weight.hpp

0 → 100644
+31 −0
Changes for include/common/rpc/random_slicing_weight.hpp: 31 added lines, 0 removed lines.
Original line number Diff line number Diff line 
/*

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

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

*/

#ifndef GKFS_RANDOM_SLICING_WEIGHT_HPP

#define GKFS_RANDOM_SLICING_WEIGHT_HPP



#include <algorithm>

#include <cmath>



namespace gkfs::rpc {



inline float

random_slicing_weight(bool perf_enabled, float throughput, float median_perf,

                      bool space_enabled, float available_gib,

                      float median_space) {

    const auto factor = [](bool enabled, float value, float median) {

        return enabled && std::isfinite(value) && value >= 0.0f &&

                               std::isfinite(median) && median > 0.0f

                       ? value / median

                       : 1.0f;

    };

    return std::clamp(

            factor(perf_enabled, throughput, median_perf) *

                    factor(space_enabled, available_gib, median_space),

            0.05f, 20.0f);

}



} // namespace gkfs::rpc



#endif // GKFS_RANDOM_SLICING_WEIGHT_HPP
 
 No newline at end of file

src/daemon/util.cpp

+75 −15
Changes for src/daemon/util.cpp: 75 added lines, 15 removed lines.
Original line number Diff line number Diff line
@@ -42,6 +42,7 @@ 
#include <daemon/env.hpp>



#include <common/rpc/rpc_util.hpp>

#include <common/rpc/random_slicing_weight.hpp>

#include <common/env.hpp>

#include <common/env_util.hpp>
@@ -57,6 +58,7 @@ 
#include <optional>

#include <thread>

#include <unistd.h>

#include <sys/statvfs.h>



using namespace std;
@@ -65,6 +67,8 @@ namespace gkfs::utils { 
namespace {



float calibrated_weight = 1.0f;

bool weight_perf = false;

bool weight_space = false;



struct CalibrationStats {

    std::string identity;
@@ -73,6 +77,9 @@ struct CalibrationStats { 
    float median_mbps = 0.0f;

    float threshold_mbps = 0.0f;

    float weight = 1.0f;

    float available_gib = 0.0f;

    float median_gib = 0.0f;

    bool space_success = false;

    bool success = false;

    bool fallback_equal_weight = true;

};
@@ -86,8 +93,25 @@ struct CalibrationResult { 
    bool success;

    float throughput_mbps;

    float weight;

    bool space_success = false;

    float available_gib = 0.0f;

};



CalibrationResult

measure_available_space(const std::filesystem::path& root,

                        CalibrationResult result) {

    struct statvfs info{};

    std::error_code ec;

    std::filesystem::create_directories(root, ec);

    if(!ec && ::statvfs(root.c_str(), &info) == 0) {

        result.space_success = true;

        result.available_gib = static_cast<float>(

                static_cast<long double>(info.f_bavail) * info.f_frsize /

                (1024.0L * 1024.0L * 1024.0L));

    }

    return result;

}



std::optional<std::size_t>

environment_integer(const char* name) {

    const auto value = gkfs::env::get_var(name);
@@ -106,8 +130,6 @@ environment_integer(const char* name) { 


CalibrationResult

benchmark_storage(const std::filesystem::path& root, int seconds) {

    if(!gkfs::env::get_var_bool(gkfs::env::RANDOM_SLICING_CALIBRATION, false))

        return {true, 100.0f, 1.0f};

    const auto path = root / ".gkfs_rs_calibration";

    constexpr std::size_t block_size = 1024 * 1024;

    constexpr std::size_t probe_size = 2ULL * 1024ULL * 1024ULL * 1024ULL;
@@ -255,6 +277,8 @@ global_calibration_weight(const std::filesystem::path& hosts_file, 
    calibration_stats.throughput_mbps = local_result.throughput_mbps;

    calibration_stats.weight = 1.0f;

    calibration_stats.success = local_result.success;

    calibration_stats.space_success = local_result.space_success;

    calibration_stats.available_gib = local_result.available_gib;

    calibration_stats.fallback_equal_weight = true;



    const auto count = calibration_daemon_count();
@@ -276,7 +300,9 @@ global_calibration_weight(const std::filesystem::path& hosts_file, 
        if(!out)

            return 1.0f;

        out << identity << ' ' << (local_result.success ? 1 : 0) << ' '

            << local_result.throughput_mbps << '\n';

            << local_result.throughput_mbps << ' '

            << (local_result.space_success ? 1 : 0) << ' '

            << local_result.available_gib << '\n';

    }

    std::filesystem::rename(temporary_path, result_path, ec);

    if(ec)
@@ -285,9 +311,11 @@ global_calibration_weight(const std::filesystem::path& hosts_file, 
    const auto deadline =

            std::chrono::steady_clock::now() + std::chrono::seconds(60);

    std::map<std::string, float> results;

    std::map<std::string, float> spaces;

    bool local_result_found = false;

    while(std::chrono::steady_clock::now() < deadline) {

        results.clear();

        spaces.clear();

        local_result_found = false;

        bool complete = true;

        for(std::size_t rank = 0; rank < *count; ++rank) {
@@ -295,7 +323,10 @@ global_calibration_weight(const std::filesystem::path& hosts_file, 
            std::string result_identity;

            int success = 0;

            float throughput = 0.0f;

            int space_success = 0;

            float available_gib = 0.0f;

            if(!(in >> result_identity >> success >> throughput) ||

               !(in >> space_success >> available_gib) ||

               result_identity.empty() ||

               results.find(result_identity) != results.end()) {

                complete = false;
@@ -305,12 +336,18 @@ global_calibration_weight(const std::filesystem::path& hosts_file, 
                results.emplace(result_identity, throughput);

            else

                results.emplace(result_identity, 0.0f);

            spaces.emplace(result_identity,

                           space_success && std::isfinite(available_gib) &&

                                           available_gib >= 0.0f

                                   ? available_gib

                                   : -1.0f);

            if(result_identity == identity)

                local_result_found = true;

        }

        if(complete && results.size() == *count && local_result_found)

            break;

        results.clear();

        spaces.clear();

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

    }

    if(results.size() != *count || !local_result_found)
@@ -321,17 +358,30 @@ global_calibration_weight(const std::filesystem::path& hosts_file, 
        if(throughput > 0.0f)

            throughputs.push_back(throughput);

    }

    if(throughputs.empty())

    std::vector<float> free_spaces;

    for(const auto& [result_identity, available] : spaces) {

        (void) result_identity;

        if(available > 0.0f)

            free_spaces.push_back(available);

    }

    if(weight_perf && throughputs.empty())

        return 1.0f;

    std::sort(throughputs.begin(), throughputs.end());

    const auto median = throughputs[throughputs.size() / 2];

    if(!std::isfinite(median) || median <= 0.0f)

    if(weight_space && free_spaces.empty())

        return 1.0f;

    std::sort(throughputs.begin(), throughputs.end());

    std::sort(free_spaces.begin(), free_spaces.end());

    const auto median =

            throughputs.empty() ? 1.0f : throughputs[throughputs.size() / 2];

    const auto median_space =

            free_spaces.empty() ? 1.0f : free_spaces[free_spaces.size() / 2];

    const auto local = results.find(identity);

    if(local == results.end() || local->second <= 0.0f)

    if(local == results.end())

        return 1.0f;

    calibration_stats.median_mbps = median;

    calibration_stats.weight = std::clamp(local->second / median, 0.05f, 20.0f);

    calibration_stats.median_gib = median_space;

    calibration_stats.weight = gkfs::rpc::random_slicing_weight(

            weight_perf, local->second > 0.0f ? local->second : -1.0f, median,

            weight_space, spaces.at(identity), median_space);

    calibration_stats.fallback_equal_weight = false;

    return calibration_stats.weight;

}
@@ -340,17 +390,22 @@ global_calibration_weight(const std::filesystem::path& hosts_file, 


void

prepare_storage_root() {

    if(!gkfs::env::get_var_bool(gkfs::env::RANDOM_SLICING_CALIBRATION, false))

    weight_perf = gkfs::env::get_var_bool(gkfs::env::RANDOM_SLICING_WEIGHT_PERF,

                                          false);

    weight_space = gkfs::env::get_var_bool(

            gkfs::env::RANDOM_SLICING_WEIGHT_SPACE, false);

    if(!weight_perf && !weight_space)

        return;



    const auto seconds = calibration_seconds();

    const auto primary = std::filesystem::path(GKFS_DATA->rootdir());

    auto final_result = benchmark_storage(primary, seconds);

    auto final_result = weight_perf ? benchmark_storage(primary, seconds)

                                    : CalibrationResult{true, 0.0f, 1.0f};

    const auto hosts_file = std::filesystem::path(GKFS_DATA->hosts_file());

    calibration_stats.selected_root = primary.string();

    calibration_stats.threshold_mbps = calibration_min_mbps();

    const auto secondary_value = gkfs::env::get_var(gkfs::env::DAEMON_2ND_ROOT);

    if(!secondary_value.empty() &&

    if(weight_perf && !secondary_value.empty() &&

       (!final_result.success ||

        final_result.throughput_mbps < calibration_min_mbps())) {

        const auto secondary = std::filesystem::path(secondary_value);
@@ -379,6 +434,9 @@ prepare_storage_root() { 
                    __func__, secondary.string(), e.what(), primary.string());

        }

    }

    if(weight_space)

        final_result =

                measure_available_space(GKFS_DATA->rootdir(), final_result);

    calibrated_weight = global_calibration_weight(hosts_file, final_result);

}
@@ -458,23 +516,25 @@ populate_hosts_file(bool expand_mode) { 
            (int) GKFS_DATA->mtime_state(), (int) GKFS_DATA->ctime_state(),

            (int) GKFS_DATA->link_cnt_state(), (int) GKFS_DATA->blocks_state(),

            getuid(), getgid());

    if(gkfs::env::get_var_bool(gkfs::env::RANDOM_SLICING_CALIBRATION, false)) {

    if(weight_perf || weight_space) {

        line_out = fmt::format("{} rs_weight={:.6f}", line_out,

                               storage_calibration_weight());

    }

    std::string calibration_comment;

    if(gkfs::env::get_var_bool(gkfs::env::RANDOM_SLICING_CALIBRATION, false)) {

    if(weight_perf || weight_space) {

        const auto status = calibration_stats.fallback_equal_weight

                                    ? "equal-weight-fallback"

                                    : "global-median";

        calibration_comment = fmt::format(

                "# GKFS_RS_CALIBRATION daemon={} root={} "

                "throughput_mbps={:.3f} median_mbps={:.3f} threshold_mbps={:.3f} "

                "available_gib={:.3f} median_gib={:.3f} perf={} space={} "

                "weight={:.6f} status={}",

                calibration_stats.identity, calibration_stats.selected_root,

                calibration_stats.throughput_mbps,

                calibration_stats.median_mbps, calibration_stats.threshold_mbps,

                calibration_stats.weight, status);

                calibration_stats.available_gib, calibration_stats.median_gib,

                weight_perf, weight_space, calibration_stats.weight, status);

    }

    // Prepend '+' marker if in expand mode

    if(expand_mode) {

tests/unit/test_random_slicing_distributor.cpp

+29 −0
Changes for tests/unit/test_random_slicing_distributor.cpp: 29 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -38,6 +38,7 @@ 
*/



#include <catch2/catch_all.hpp>

#include <common/rpc/random_slicing_weight.hpp>

#include <common/hostfile_management.hpp>

#include <common/rpc/cutshift_sorted.hpp>

#include <common/rpc/random_slicing_distributor.hpp>
@@ -209,6 +210,34 @@ TEST_CASE("RandomSlicingDistributor honors weights", 
    REQUIRE(partitions[1].coverage() == Catch::Approx(0.75f));

}



TEST_CASE("Random slicing combines independent performance and space weights",

          "[common][distributor][random_slicing][weights]") {

    using gkfs::rpc::random_slicing_weight;

    REQUIRE(random_slicing_weight(false, 2.0f, 1.0f, false, 0.25f,

                                  1.0f) == Catch::Approx(1.0f));

    REQUIRE(random_slicing_weight(true, 2.0f, 1.0f, false, 0.25f,

                                  1.0f) == Catch::Approx(2.0f));

    REQUIRE(random_slicing_weight(false, 2.0f, 1.0f, true, 0.25f,

                                  1.0f) == Catch::Approx(0.25f));

    REQUIRE(random_slicing_weight(true, 2.0f, 1.0f, true, 0.25f,

                                  1.0f) == Catch::Approx(0.5f));

    REQUIRE(random_slicing_weight(false, 1.0f, 1.0f, true, 0.0f,

                                  1.0f) == Catch::Approx(0.05f));

    REQUIRE(random_slicing_weight(false, 1.0f, 1.0f, true, -1.0f,

                                  1.0f) == Catch::Approx(1.0f));

    REQUIRE(random_slicing_weight(true, -1.0f, 1.0f, true, 0.25f,

                                  1.0f) == Catch::Approx(0.25f));

    REQUIRE(random_slicing_weight(true, 100.0f, 1.0f, true, 2.0f,

                                  1.0f) == Catch::Approx(20.0f));



    gkfs::rpc::RandomSlicingDistributor distributor(

            0, 2, {random_slicing_weight(true, 2.0f, 1.0f, true, 0.25f,

                                       1.0f),

                   1.0f});

    REQUIRE(distributor.get_partitions()[0].coverage() ==

            Catch::Approx(1.0f / 3.0f));

}



TEST_CASE("Hostfile RS weights are sorted and tolerate invalid values",

          "[common][hostfile][random_slicing]") {

    const auto path = std::filesystem::temp_directory_path() /