feat(rpc): enable guided distributor via factory and environment variable (35167706) · Commits · hpc / gekkofs

include/common/env.hpp

+2 −0

Changes for include/common/env.hpp: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -56,6 +56,8 @@ static constexpr auto METRICS_AGGREGATOR = ADD_PREFIX("METRICS_AGGREGATOR");

		static constexpr auto DISTRIBUTION_STRATEGY =
		ADD_PREFIX("DISTRIBUTION_STRATEGY");
		static constexpr auto GUIDED_DISTRIBUTION_PATH =
		ADD_PREFIX("GUIDED_DISTRIBUTION_PATH");
		static constexpr auto RANDOM_SLICING_CUTSHIFT =
		ADD_PREFIX("RANDOM_SLICING_CUTSHIFT");
		static constexpr auto RANDOM_SLICING_WEIGHT_PERF =

include/common/rpc/distribution_config.hpp

+6 −0

Changes for include/common/rpc/distribution_config.hpp: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -88,6 +88,12 @@ public:
		return strategy_to_string(strategy_);
		}

		/// Check if guided distribution is active
		bool
		is_guided() const {
		return strategy_ == DistributionStrategy::Guided;
		}

		/// Check if random slicing is active
		bool
		is_random_slicing() const {

include/common/rpc/distributor.hpp

+1 −1

Changes for include/common/rpc/distributor.hpp: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -242,7 +242,7 @@ private:
		std::hash<std::string> str_hash;
		std::unordered_map<std::string, std::pair<IntervalSet, unsigned int>>
		map_interval;
		std::vector<std::string> prefix_list; // Should not be very long
		std::vector<std::pair<std::string, host_t>> prefix_list; // Should not be very long
		bool
		init_guided();

include/common/rpc/distributor_factory.hpp

+1 −0

Changes for include/common/rpc/distributor_factory.hpp: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -70,6 +70,7 @@ namespace rpc {
		/// Supported distribution strategies
		enum class DistributionStrategy {
		SimpleHash, // current default: modulo-based placement
		Guided, // guided placement from mapping file
		RandomSlicing, // new: interval-based placement
		LocalOnly, // all data on local node
		Forwarder // data forwarded to a specific host

src/common/rpc/distributor.cpp

+56 −9

Changes for src/common/rpc/distributor.cpp: 56 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		#include <common/env.hpp>
		#include <cstdlib>
		/*
		Copyright 2018-2026, Barcelona Supercomputing Center (BSC), Spain
		Copyright 2015-2026, Johannes Gutenberg Universitaet Mainz, Germany
		@@ -218,7 +220,18 @@ GuidedDistributor::init_guided() {
		chunkid_t chunk_id;
		string path;
		std::ifstream mapfile;
		mapfile.open(GKFS_USE_GUIDED_DISTRIBUTION_PATH);
		const char* env_path = std::getenv(gkfs::env::GUIDED_DISTRIBUTION_PATH);
		if(env_path == nullptr \|\| env_path[0] == '\0') {
		env_path = std::getenv("LIBGKFS_GUIDED_DISTRIBUTION_PATH");
		}
		if(env_path == nullptr \|\| env_path[0] == '\0') {
		env_path = std::getenv("GKFS_GUIDED_PATH");
		}
		const std::string filepath = (env_path != nullptr && env_path[0] != '\0')
		? std::string(env_path)
		: std::string(GKFS_USE_GUIDED_DISTRIBUTION_PATH);

		mapfile.open(filepath);
		if((mapfile.rdstate() & std::ifstream::failbit) != 0)
		return false; // If it fails, the mapping will be as the SimpleHash

		@@ -226,10 +239,9 @@ GuidedDistributor::init_guided() {
		// We need destination+1, as 0 has an special meaning in the interval
		// map.
		if(path.size() > 0 and path[0] == '#') {
		// Path that has this prefix will have metadata and data in the same
		// place i.e. #/mdtest-hard/ 10 10 chunk_id and destination_host
		// are not used
		prefix_list.emplace_back(path.substr(1, path.size()));
		// Path that has this prefix will direct data and metadata to destination_host
		// if valid (< hosts_size_), or co-locate on hashed host if destination_host >= hosts_size_
		prefix_list.emplace_back(path.substr(1, path.size()), destination_host);
		continue;
		}

		@@ -254,12 +266,10 @@ GuidedDistributor::GuidedDistributor() {

		GuidedDistributor::GuidedDistributor(host_t localhost,
		unsigned int hosts_size) {
		if(hosts_size_ != hosts_size) {
		hosts_size_ = hosts_size;
		localhost_ = localhost;
		hosts_size_ = hosts_size;
		all_hosts_ = std::vector<unsigned int>(hosts_size);
		::iota(all_hosts_.begin(), all_hosts_.end(), 0);
		}
		init_guided();
		}

		@@ -302,12 +312,26 @@ GuidedDistributor::locate_data(const string& path, const chunkid_t& chnk_id,
		}
		}

		for(auto const& prefix : prefix_list) {
		for(auto const& prefix_rule : prefix_list) {
		const auto& prefix = prefix_rule.first;
		if(0 == path.compare(0, min(prefix.length(), path.length()), prefix, 0,
		min(prefix.length(), path.length()))) {
		if(hosts_size_ > 0 && prefix_rule.second < hosts_size_) {
		return prefix_rule.second;
		}
		return str_hash(path) % hosts_size_;
		}
		}

		for(auto const& [mapped_path, entry] : map_interval) {
		if(path.length() > mapped_path.length() &&
		path.compare(0, mapped_path.length(), mapped_path) == 0 &&
		(mapped_path.back() == '/' \|\| path[mapped_path.length()] == '/')) {
		if(entry.first.IsInsideInterval(chnk_id)) {
		return (entry.second - 1);
		}
		}
		}

		auto locate = path + ::to_string(chnk_id);
		return (str_hash(locate) + num_copy) % hosts_size_;
		@@ -316,6 +340,29 @@ GuidedDistributor::locate_data(const string& path, const chunkid_t& chnk_id,
		host_t
		GuidedDistributor::locate_file_metadata(const string& path,
		const int num_copy) const {
		for(auto const& prefix_rule : prefix_list) {
		const auto& prefix = prefix_rule.first;
		if(0 == path.compare(0, min(prefix.length(), path.length()), prefix, 0,
		min(prefix.length(), path.length()))) {
		if(hosts_size_ > 0 && prefix_rule.second < hosts_size_) {
		return prefix_rule.second;
		}
		return (str_hash(path) + num_copy) % hosts_size_;
		}
		}

		auto it = map_interval.find(path);
		if(it != map_interval.end()) {
		return (it->second.second - 1);
		}
		for(auto const& [mapped_path, entry] : map_interval) {
		if(path.length() > mapped_path.length() &&
		path.compare(0, mapped_path.length(), mapped_path) == 0 &&
		(mapped_path.back() == '/' \|\| path[mapped_path.length()] == '/')) {
		return (entry.second - 1);
		}
		}

		return (str_hash(path) + num_copy) % hosts_size_;
		}