Commit 14bdda95 authored by Ramon Nou's avatar Ramon Nou
Browse files

Merge branch 'rnou/opt2' into 'master' 

Weight Random Slicing by storage performance and available space

In MN5 runs, we hit a out of space issue normally in mdtest-hard-write phase.
This is a big stopper, a debug analysis show that eventually some of the space is full (due to other users filling the NVMe outside the designed job directory and not deleting the data)

This MR modifies the random slicing calibration (perf) with two weighting (multi ON supported)

```
* GKFS_RANDOM_SLICING_WEIGHT_SPACE=ON/OFF
* GKFS_RANDOM_SLICING_WEIGHT_PERF=ON/OFF
* GKFS_DISTRIBUTION_STRATEGY=random_slicing
```

See merge request !322
parents 8b0490b4 0e997ade

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() /