Enhance client endpoint lookups with retry logic to improve test reliability.... (f70d54e5) · Commits · hpc / gekkofs

examples/gfind/sfind.cpp

+12 −4

Original line number	Diff line number	Diff line
		@@ -302,11 +302,15 @@ worker_routine(void* arg) {

		if(n > 0 && entries && !data->opt->just_count) {
		char* ptr = reinterpret_cast<char*>(entries);
		int bytes_processed = 0;
		size_t bytes_processed = 0;
		while(bytes_processed < n) {
		struct dirent_extended* temp =
		reinterpret_cast<struct dirent_extended*>(ptr);
		if(temp->d_reclen == 0)
		const auto remaining =
		static_cast<size_t>(n) - bytes_processed;
		if(temp->d_reclen <
		offsetof(struct dirent_extended, d_name) + 1 \|\|
		temp->d_reclen > remaining)
		break;

		local_found++;
		@@ -360,11 +364,15 @@ worker_routine(void* arg) {
		}

		char* ptr = reinterpret_cast<char*>(entries);
		int bytes_processed = 0;
		size_t bytes_processed = 0;
		while(bytes_processed < n) {
		struct dirent_extended* temp =
		reinterpret_cast<struct dirent_extended*>(ptr);
		if(temp->d_reclen == 0)
		const auto remaining =
		static_cast<size_t>(n) - bytes_processed;
		if(temp->d_reclen <
		offsetof(struct dirent_extended, d_name) + 1 \|\|
		temp->d_reclen > remaining)
		break;

		if(temp->d_type != 1) {

include/daemon/classes/fs_data.hpp

+13 −0

Original line number	Diff line number	Diff line
		@@ -139,6 +139,10 @@ private:
		// indicates for clients: try again. Is set to true when redist is running
		bool maintenance_mode_ = false;
		ABT_mutex maintenance_mode_mutex_;
		bool mutate_start_active_ = false;
		int mutate_old_server_conf_ = 0;
		int mutate_new_server_conf_ = 0;
		std::string mutate_hosts_file_;
		// redist_running_ indicates to client that redistribution is running
		std::atomic<bool> redist_running_{false};

		@@ -370,6 +374,15 @@ public:
		void
		maintenance_mode(bool maintenance_mode);

		// Returns true when this request is a duplicate of the active mutation,
		// false when a new mutation was registered, and throws for a conflict.
		bool
		begin_mutate_start(int old_server_conf, int new_server_conf,
		const std::string& hosts_file);

		void
		end_mutate_start();

		bool
		redist_running() const;

src/client/rpc/forward_malleability.cpp

+52 −4

Original line number	Diff line number	Diff line
		@@ -31,12 +31,33 @@

		#include <set>
		#include <cerrno>
		#include <chrono>
		#include <cstdlib>
		#include <string>
		#include <thread>
		#include <vector>

		namespace {

		thallium::endpoint
		lookup_mutate_endpoint(const std::string& uri) {
		constexpr unsigned int max_attempts = 10;
		for(unsigned int attempt = 0; attempt < max_attempts; ++attempt) {
		try {
		return CTX->rpc_engine()->lookup(uri);
		} catch(const std::exception& ex) {
		if(attempt + 1 == max_attempts) {
		throw;
		}
		LOG(DEBUG,
		"Lookup of mutate endpoint '{}' failed (attempt {}/{}): {}",
		uri, attempt + 1, max_attempts, ex.what());
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
		}
		}
		throw std::runtime_error("unreachable endpoint lookup failure");
		}

		std::vector<std::pair<std::string, thallium::endpoint>>
		mutate_endpoints_from_loaded_hosts() {
		std::vector<std::pair<std::string, thallium::endpoint>> endpoints;
		@@ -74,7 +95,7 @@ mutate_endpoints_from_markers(const std::string& hostfile,
		try {
		LOG(DEBUG, "Looking up mutate endpoint '{}'", entry.uri);
		endpoints.emplace_back(entry.uri,
		CTX->rpc_engine()->lookup(entry.uri));
		lookup_mutate_endpoint(entry.uri));
		} catch(const std::exception& ex) {
		LOG(ERROR, "Failed to lookup mutate endpoint '{}': {}", entry.uri,
		ex.what());
		@@ -109,7 +130,7 @@ mutate_removed_endpoints_from_markers(const std::string& hostfile) {
		try {
		LOG(DEBUG, "Looking up removed mutate endpoint '{}'", entry.uri);
		endpoints.emplace_back(entry.uri,
		CTX->rpc_engine()->lookup(entry.uri));
		lookup_mutate_endpoint(entry.uri));
		} catch(const std::exception& ex) {
		LOG(ERROR, "Failed to lookup removed mutate endpoint '{}': {}",
		entry.uri, ex.what());
		@@ -160,10 +181,33 @@ forward_mutate_start(int old_server_conf, int new_server_conf,
		LOG(INFO, "{}() enter", __func__);
		auto targets = mutate_endpoints(&new_hosts_file, true);

		if(targets.empty()) {
		std::size_t expected_targets = targets.size();
		try {
		const auto markers =
		gkfs::malleable::parse_hostfile_markers(new_hosts_file);
		std::set<std::string> expected_uris;
		for(const auto& entry : markers.active) {
		expected_uris.insert(entry.uri);
		}
		for(const auto& entry : markers.adding) {
		expected_uris.insert(entry.uri);
		}
		for(const auto& entry : markers.removing) {
		expected_uris.insert(entry.uri);
		}
		expected_targets = expected_uris.size();
		} catch(const std::exception& ex) {
		LOG(ERROR, "Failed to validate mutate targets from '{}': {}",
		new_hosts_file, ex.what());
		return EINVAL;
		}

		if(targets.empty() \|\| targets.size() != expected_targets) {
		LOG(ERROR, "No mutate targets found for hosts file '{}'",
		new_hosts_file);
		return EINVAL;
		LOG(ERROR, "Expected {} mutate targets, resolved {}", expected_targets,
		targets.size());
		return EBUSY;
		}

		auto err = 0;
		@@ -201,8 +245,12 @@ forward_mutate_start(int old_server_conf, int new_server_conf,
		try {
		gkfs::rpc::rpc_err_out_t out = waiters[i].wait();
		if(out.err != 0) {
		LOG(ERROR, "Mutate start failed on target '{}' with error {}",
		waiter_targets[i], out.err);
		if(err == 0) {
		err = out.err;
		}
		}
		} catch(const std::exception& ex) {
		LOG(ERROR, "RPC wait failed for target {}: {}", waiter_targets[i],
		ex.what());

src/daemon/classes/fs_data.cpp

+31 −0

Original line number	Diff line number	Diff line
		@@ -404,6 +404,37 @@ FsData::maintenance_mode(bool maintenance_mode) {
		ABT_mutex_unlock(maintenance_mode_mutex_);
		}

		bool
		FsData::begin_mutate_start(int old_server_conf, int new_server_conf,
		const std::string& hosts_file) {
		ABT_mutex_lock(maintenance_mode_mutex_);
		if(mutate_start_active_) {
		const bool duplicate = mutate_old_server_conf_ == old_server_conf &&
		mutate_new_server_conf_ == new_server_conf &&
		mutate_hosts_file_ == hosts_file;
		ABT_mutex_unlock(maintenance_mode_mutex_);
		if(duplicate) {
		return true;
		}
		throw std::runtime_error(
		"A different mutate operation is already active");
		}
		mutate_start_active_ = true;
		mutate_old_server_conf_ = old_server_conf;
		mutate_new_server_conf_ = new_server_conf;
		mutate_hosts_file_ = hosts_file;
		ABT_mutex_unlock(maintenance_mode_mutex_);
		return false;
		}

		void
		FsData::end_mutate_start() {
		ABT_mutex_lock(maintenance_mode_mutex_);
		mutate_start_active_ = false;
		mutate_hosts_file_.clear();
		ABT_mutex_unlock(maintenance_mode_mutex_);
		}

		bool
		FsData::redist_running() const {
		return redist_running_.load();

src/daemon/handler/srv_malleability.cpp

+14 −0

Original line number	Diff line number	Diff line
		@@ -59,12 +59,22 @@ rpc_srv_mutate_start(const tl::request& req,
		const gkfs::rpc::rpc_mutate_start_in_t& in) {
		gkfs::rpc::rpc_err_out_t out;
		bool entered_maintenance = false;
		bool duplicate_request = false;
		bool registered_mutate = false;

		GKFS_DATA->spdlogger()->debug(
		"{}() Got RPC with old conf '{}' new conf '{}' new_hosts_file '{}'",
		__func__, in.old_server_conf, in.new_server_conf,
		in.new_hosts_file);
		try {
		duplicate_request = GKFS_DATA->begin_mutate_start(
		in.old_server_conf, in.new_server_conf, in.new_hosts_file);
		if(duplicate_request) {
		out.err = 0;
		gkfs::utils::safe_respond(req, out);
		return;
		}
		registered_mutate = true;
		GKFS_DATA->maintenance_mode(true);
		entered_maintenance = true;
		GKFS_DATA->malleable_manager()->mutate_start(
		@@ -76,6 +86,9 @@ rpc_srv_mutate_start(const tl::request& req,
		if(entered_maintenance) {
		GKFS_DATA->maintenance_mode(false);
		}
		if(registered_mutate) {
		GKFS_DATA->end_mutate_start();
		}
		out.err = -1;
		}

		@@ -108,6 +121,7 @@ rpc_srv_mutate_finalize(const tl::request& req) {
		try {
		GKFS_DATA->maintenance_mode(false);
		GKFS_DATA->keep_hosts_file(true);
		GKFS_DATA->end_mutate_start();
		out.err = 0;
		} catch(const std::exception& e) {
		GKFS_DATA->spdlogger()->error("{}() Failed to finalize mutate: '{}'",