Persist mutation recovery checkpoints (35bb562d) · Commits · hpc / gekkofs

include/daemon/classes/fs_data.hpp

+8 −0

Changes for include/daemon/classes/fs_data.hpp: 8 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -477,6 +477,14 @@ public:
		void
		mutation_committed();

		void
		restore_mutation_recovery_state(uint64_t old_generation,
		uint64_t new_generation,
		mutation_phase phase);

		void
		restore_placement_generation(uint64_t generation);

		bool
		redist_running() const;

include/daemon/malleability/malleable_manager.hpp

+7 −0

Changes for include/daemon/malleability/malleable_manager.hpp: 7 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -88,6 +88,13 @@ public:
		void
		wait_for_worker();

		/**
		* Load the last migration checkpoint. Interrupted work is restored as
		* failed and is never resumed or finalized automatically.
		*/
		void
		load_migration_checkpoint();

		void
		mutate_start(int old_server_conf, int new_server_conf,
		const std::string& new_hosts_file);

src/daemon/classes/fs_data.cpp

+19 −0

Changes for src/daemon/classes/fs_data.cpp: 19 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -629,6 +629,25 @@ FsData::mutation_committed() {
		ABT_mutex_unlock(maintenance_mode_mutex_);
		}

		void
		FsData::restore_mutation_recovery_state(const uint64_t old_generation,
		const uint64_t new_generation,
		const mutation_phase phase) {
		ABT_mutex_lock(maintenance_mode_mutex_);
		mutate_old_placement_generation_ = old_generation;
		mutate_new_placement_generation_ = new_generation;
		mutation_phase_ = phase;
		mutate_start_active_ = phase != mutation_phase::idle;
		ABT_mutex_unlock(maintenance_mode_mutex_);
		}

		void
		FsData::restore_placement_generation(const uint64_t generation) {
		ABT_mutex_lock(maintenance_mode_mutex_);
		placement_generation_ = generation;
		ABT_mutex_unlock(maintenance_mode_mutex_);
		}

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

src/daemon/daemon.cpp

+1 −0

Changes for src/daemon/daemon.cpp: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -779,6 +779,7 @@ init_environment() {
		auto malleable_manager =
		std::make_shared<gkfs::malleable::MalleableManager>();
		GKFS_DATA->malleable_manager(malleable_manager);
		malleable_manager->load_migration_checkpoint();

		} catch(const std::exception& e) {
		GKFS_DATA->spdlogger()->error(

src/daemon/malleability/malleable_manager.cpp

+91 −5

Changes for src/daemon/malleability/malleable_manager.cpp: 91 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -78,7 +78,18 @@ std::string
		migration_checkpoint_path() {
		const auto hostfile = GKFS_DATA->hosts_file();
		const auto directory = hostfile + ".ready";
		return fmt::format("{}/migration_{}.checkpoint", directory, getpid());
		std::string identity = gkfs::rpc::get_my_hostname(true);
		const auto suffix = GKFS_DATA->rootdir_suffix();
		if(!suffix.empty()) {
		identity += "_" + suffix;
		}
		for(auto& character : identity) {
		if(!std::isalnum(static_cast<unsigned char>(character)) &&
		character != '_' && character != '-') {
		character = '_';
		}
		}
		return fmt::format("{}/migration_{}.checkpoint", directory, identity);
		}

		void
		@@ -96,12 +107,17 @@ write_migration_checkpoint(const std::string& state, size_t jobs_total,
		const auto temporary = fmt::format("{}.tmp.{}", path,
		static_cast<long long>(getpid()));
		const auto content = fmt::format(
		"format=1\nstate={}\npid={}\nhostfile={}\njobs_total={}\n"
		"format=2\nstate={}\nphase={}\npid={}\nhostfile={}\n"
		"placement_old_generation={}\nplacement_new_generation={}\n"
		"placement_generation={}\njobs_total={}\n"
		"jobs_completed={}\njobs_succeeded={}\njobs_failed={}\n"
		"bytes_transferred={}\nerror={}\n",
		state, getpid(), GKFS_DATA->hosts_file(), jobs_total,
		jobs_completed, jobs_succeeded, jobs_failed, bytes_transferred,
		error);
		state, static_cast<uint8_t>(GKFS_DATA->mutation_state()),
		getpid(), GKFS_DATA->hosts_file(),
		GKFS_DATA->mutate_old_placement_generation(),
		GKFS_DATA->mutate_new_placement_generation(),
		GKFS_DATA->placement_generation(), jobs_total, jobs_completed,
		jobs_succeeded, jobs_failed, bytes_transferred, error);
		gkfs::utils::unique_fd fd(
		open(temporary.c_str(), O_WRONLY \| O_CREAT \| O_TRUNC, 0640));
		if(!fd) {
		@@ -180,6 +196,76 @@ MalleableManager::wait_for_worker() {
		join_redist_thread();
		}

		void
		MalleableManager::load_migration_checkpoint() {
		if(!GKFS_DATA->migration_checkpoint()) {
		return;
		}

		const auto path = migration_checkpoint_path();
		std::ifstream input(path);
		if(!input.is_open()) {
		return;
		}

		std::unordered_map<std::string, std::string> fields;
		std::string line;
		while(std::getline(input, line)) {
		const auto separator = line.find('=');
		if(separator != std::string::npos) {
		fields.emplace(line.substr(0, separator),
		line.substr(separator + 1));
		}
		}

		const auto state_it = fields.find("state");
		if(state_it == fields.end()) {
		return;
		}

		try {
		const auto old_generation =
		std::stoull(fields.count("placement_old_generation")
		? fields.at("placement_old_generation")
		: "0");
		const auto new_generation =
		std::stoull(fields.count("placement_new_generation")
		? fields.at("placement_new_generation")
		: "0");
		const auto committed_generation =
		std::stoull(fields.count("placement_generation")
		? fields.at("placement_generation")
		: "0");
		const auto phase = static_cast<gkfs::daemon::mutation_phase>(
		std::stoul(fields.count("phase") ? fields.at("phase") : "0"));

		if(state_it->second == "complete") {
		GKFS_DATA->restore_placement_generation(committed_generation);
		return;
		}

		if(state_it->second == "running" \|\| state_it->second == "failed" \|\|
		state_it->second == "complete") {
		GKFS_DATA->restore_mutation_recovery_state(
		old_generation, new_generation,
		phase == gkfs::daemon::mutation_phase::failed
		? phase
		: gkfs::daemon::mutation_phase::failed);
		GKFS_DATA->redist_failed(true);
		return;
		}

		GKFS_DATA->restore_mutation_recovery_state(
		old_generation, new_generation,
		gkfs::daemon::mutation_phase::failed);
		GKFS_DATA->redist_failed(true);
		} catch(const std::exception& e) {
		GKFS_DATA->spdlogger()->warn(
		"{}() Ignoring malformed migration checkpoint '{}': {}",
		__func__, path, e.what());
		}
		}

		static string
		host_key(pair<string, string> host) {
		if(auto idx = host.first.rfind("#"); idx != string::npos) {