Add calibration statistics to random slicing hostfile (ca06e25a) · Commits · hpc / gekkofs

include/common/env.hpp

+4 −0

Changes for include/common/env.hpp: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -64,6 +64,10 @@ static constexpr auto RANDOM_SLICING_CALIBRATION_SECONDS =
		ADD_PREFIX("RANDOM_SLICING_CALIBRATION_SECONDS");
		static constexpr auto RANDOM_SLICING_CALIBRATION_MIN_MBPS =
		ADD_PREFIX("RANDOM_SLICING_CALIBRATION_MIN_MBPS");
		static constexpr auto RANDOM_SLICING_CALIBRATION_DAEMON_COUNT =
		ADD_PREFIX("RANDOM_SLICING_CALIBRATION_DAEMON_COUNT");
		static constexpr auto RANDOM_SLICING_CALIBRATION_DAEMON_ID =
		ADD_PREFIX("RANDOM_SLICING_CALIBRATION_DAEMON_ID");
		static constexpr auto EXPAND_ON_DEMAND = ADD_PREFIX("EXPAND_ON_DEMAND");
		static constexpr auto NUM_REPL = ADD_PREFIX("NUM_REPL");
		static constexpr auto CREATE_CHECK_PARENTS = ADD_PREFIX("CREATE_CHECK_PARENTS");

src/daemon/util.cpp

+183 −9

Changes for src/daemon/util.cpp: 183 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -53,6 +53,7 @@
		#include <limits>
		#include <cmath>
		#include <filesystem>
		#include <optional>
		#include <thread>
		#include <unistd.h>

		@@ -64,12 +65,44 @@ namespace {

		float calibrated_weight = 1.0f;

		struct CalibrationStats {
		std::string identity;
		std::string selected_root;
		float throughput_mbps = 0.0f;
		float median_mbps = 0.0f;
		float threshold_mbps = 0.0f;
		float weight = 1.0f;
		bool success = false;
		bool fallback_equal_weight = true;
		};

		CalibrationStats calibration_stats;

		std::string
		calibration_identity();

		struct CalibrationResult {
		bool success;
		float throughput_mbps;
		float weight;
		};

		std::optional<std::size_t>
		environment_integer(const char* name) {
		const auto value = gkfs::env::get_var(name);
		if(value.empty())
		return std::nullopt;
		try {
		std::size_t consumed = 0;
		const auto parsed = std::stoull(value, &consumed, 10);
		if(consumed != value.size())
		return std::nullopt;
		return static_cast<std::size_t>(parsed);
		} catch(const std::exception&) {
		return std::nullopt;
		}
		}

		CalibrationResult
		benchmark_storage(const std::filesystem::path& root, int seconds) {
		if(!gkfs::env::get_var_bool(gkfs::env::RANDOM_SLICING_CALIBRATION, false))
		@@ -145,6 +178,127 @@ calibration_min_mbps() {
		}
		}

		std::optional<std::size_t>
		calibration_daemon_count() {
		if(const auto count = environment_integer(
		gkfs::env::RANDOM_SLICING_CALIBRATION_DAEMON_COUNT))
		return *count;
		if(const auto count = environment_integer("SLURM_NTASKS"))
		return *count;
		if(const auto count = environment_integer("SLURM_JOB_NUM_NODES"))
		return *count;
		return std::nullopt;
		}

		std::optional<std::size_t>
		calibration_daemon_id() {
		if(const auto id = environment_integer(
		gkfs::env::RANDOM_SLICING_CALIBRATION_DAEMON_ID))
		return *id;
		if(const auto id = environment_integer("SLURM_PROCID"))
		return *id;
		if(const auto id = environment_integer("PMI_RANK"))
		return *id;
		return std::nullopt;
		}

		std::string
		calibration_identity() {
		const auto hostname = gkfs::rpc::get_my_hostname(true);
		return GKFS_DATA->rootdir_suffix().empty()
		? hostname
		: fmt::format("{}#{}", hostname,
		GKFS_DATA->rootdir_suffix());
		}

		float
		global_calibration_weight(const std::filesystem::path& hosts_file,
		const CalibrationResult& local_result) {
		calibration_stats.identity = calibration_identity();
		calibration_stats.throughput_mbps = local_result.throughput_mbps;
		calibration_stats.weight = 1.0f;
		calibration_stats.success = local_result.success;
		calibration_stats.fallback_equal_weight = true;

		const auto count = calibration_daemon_count();
		const auto id = calibration_daemon_id();
		if(!count \|\| count == 0 \|\| !id \|\| id >= *count)
		return 1.0f;

		const auto directory =
		std::filesystem::path(hosts_file.string() + ".calibration");
		std::error_code ec;
		std::filesystem::create_directories(directory, ec);
		if(ec)
		return 1.0f;
		const auto identity = calibration_stats.identity;
		const auto result_path = directory / ("result_" + std::to_string(*id));
		const auto temporary_path = result_path.string() + ".tmp";
		{
		std::ofstream out(temporary_path, std::ios::trunc);
		if(!out)
		return 1.0f;
		out << identity << ' ' << (local_result.success ? 1 : 0) << ' '
		<< local_result.throughput_mbps << '\n';
		}
		std::filesystem::rename(temporary_path, result_path, ec);
		if(ec)
		return 1.0f;

		const auto deadline =
		std::chrono::steady_clock::now() + std::chrono::seconds(60);
		std::map<std::string, float> results;
		bool local_result_found = false;
		while(std::chrono::steady_clock::now() < deadline) {
		results.clear();
		local_result_found = false;
		bool complete = true;
		for(std::size_t rank = 0; rank < *count; ++rank) {
		std::ifstream in(directory / ("result_" + std::to_string(rank)));
		std::string result_identity;
		int success = 0;
		float throughput = 0.0f;
		if(!(in >> result_identity >> success >> throughput) \|\|
		result_identity.empty() \|\|
		results.find(result_identity) != results.end()) {
		complete = false;
		break;
		}
		if(success && std::isfinite(throughput) && throughput > 0.0f)
		results.emplace(result_identity, throughput);
		else
		results.emplace(result_identity, 0.0f);
		if(result_identity == identity)
		local_result_found = true;
		}
		if(complete && results.size() == *count && local_result_found)
		break;
		results.clear();
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
		}
		if(results.size() != *count \|\| !local_result_found)
		return 1.0f;
		std::vector<float> throughputs;
		for(const auto& [result_identity, throughput] : results) {
		(void) result_identity;
		if(throughput > 0.0f)
		throughputs.push_back(throughput);
		}
		if(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)
		return 1.0f;
		const auto local = results.find(identity);
		if(local == results.end() \|\| local->second <= 0.0f)
		return 1.0f;
		calibration_stats.median_mbps = median;
		calibration_stats.weight = std::clamp(local->second / median, 0.05f, 20.0f);
		calibration_stats.fallback_equal_weight = false;
		return calibration_stats.weight;
		}

		} // namespace

		void
		@@ -154,11 +308,14 @@ prepare_storage_root() {

		const auto seconds = calibration_seconds();
		const auto primary = std::filesystem::path(GKFS_DATA->rootdir());
		const auto primary_result = benchmark_storage(primary, seconds);
		auto final_result = benchmark_storage(primary, seconds);
		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() &&
		(!primary_result.success \|\|
		primary_result.throughput_mbps < calibration_min_mbps())) {
		(!final_result.success \|\|
		final_result.throughput_mbps < calibration_min_mbps())) {
		const auto secondary = std::filesystem::path(secondary_value);
		try {
		std::filesystem::create_directories(secondary);
		@@ -166,25 +323,26 @@ prepare_storage_root() {
		"{}() Primary storage calibration {} ({:.2f} MiB/s) "
		"requires fallback; switching root to '{}'",
		__func__,
		primary_result.success ? "below threshold" : "failed",
		primary_result.throughput_mbps, secondary.string());
		final_result.success ? "below threshold" : "failed",
		final_result.throughput_mbps, secondary.string());
		GKFS_DATA->rootdir(secondary.string());
		const auto secondary_result = benchmark_storage(secondary, seconds);
		if(secondary_result.success) {
		calibrated_weight = secondary_result.weight;
		return;
		}
		final_result = secondary_result;
		calibration_stats.selected_root = secondary.string();
		} else {
		GKFS_DATA->rootdir(primary.string());
		GKFS_DATA->spdlogger()->warn(
		"{}() Secondary root '{}' calibration failed; keeping '{}'",
		__func__, secondary.string(), primary.string());
		}
		} catch(const std::exception& e) {
		GKFS_DATA->spdlogger()->warn(
		"{}() Failed to use secondary root '{}': {}; keeping '{}'",
		__func__, secondary.string(), e.what(), primary.string());
		}
		}
		calibrated_weight = primary_result.success ? primary_result.weight : 1.0f;
		calibrated_weight = global_calibration_weight(hosts_file, final_result);
		}

		float
		@@ -267,6 +425,20 @@ populate_hosts_file(bool expand_mode) {
		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)) {
		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} "
		"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);
		}
		// Prepend '+' marker if in expand mode
		if(expand_mode) {
		line_out = "+" + line_out;
		@@ -286,6 +458,8 @@ populate_hosts_file(bool expand_mode) {
		hosts_file, strerror(errno)));
		}
		lfstream << line_out << std::endl;
		if(!calibration_comment.empty())
		lfstream << calibration_comment << std::endl;
		if(!lfstream) {
		throw runtime_error(
		fmt::format("Failed to write on hosts file '{}': {}",