rs3 - expand = shrink (f1204c8f) · Commits · hpc / gekkofs

include/common/rpc/data_migration_executor.hpp

+7 −2

Original line number	Diff line number	Diff line
		@@ -38,15 +38,20 @@ enum class MigrationStatus { Success, PartialFailure, AllFailed, Cancelled };
		/// Execute a migration plan
		class DataMigrationExecutor {
		public:
		/// Per-job migration function type: returns 0 on success, -1 on failure
		using JobHandler = std::function<int(MigrationJob&)>;

		/// Execute all migration jobs with optional progress callback
		MigrationStatus
		execute(std::vector<MigrationJob>& jobs,
		MigrationProgressCallback progress = nullptr);
		MigrationProgressCallback progress = nullptr,
		JobHandler handler = nullptr);

		/// Execute migration jobs in batches of given size
		MigrationStatus
		execute_batched(std::vector<MigrationJob>& jobs, size_t batch_size,
		MigrationProgressCallback progress = nullptr);
		MigrationProgressCallback progress = nullptr,
		JobHandler handler = nullptr);

		/// Get total bytes migrated (for accounting)
		size_t

+2 −1

Original line number	Diff line number	Diff line
		@@ -450,10 +450,11 @@ struct rpc_proxy_daemon_read_in_t {
		struct rpc_expand_start_in_t {
		uint32_t old_server_conf;
		uint32_t new_server_conf;
		std::string new_hosts_file;
		template <class Archive>
		void
		serialize(Archive& ar) {
		ar(old_server_conf, new_server_conf);
		ar(old_server_conf, new_server_conf, new_hosts_file);
		}
		};

+25 −1

Original line number	Diff line number	Diff line
		@@ -39,6 +39,8 @@
		#define GEKKOFS_DAEMON_MALLEABLE_MANAGER_HPP

		#include <daemon/daemon.hpp>
		#include <common/rpc/data_migrator.hpp>
		#include <common/rpc/data_migration_executor.hpp>

		namespace gkfs::malleable {

		@@ -46,6 +48,9 @@ class MalleableManager {
		private:
		ABT_thread redist_thread_;

		// Tracks old hosts_size before expansion/shrink
		unsigned int old_hosts_size_{0};

		// TODO next 3 functions are mostly copy paste from preload_util. FIX

		std::vector<std::pair<std::string, std::string>>
		@@ -65,12 +70,31 @@ private:
		void
		redistribute_data();

		int
		execute_migrations(const std::vector<gkfs::rpc::MigrationJob>& jobs);

		/// Migration job handler: called by DataMigrationExecutor for each job
		int
		do_migration(gkfs::rpc::MigrationJob& job);

		static void
		expand_abt(void* _arg);

		/// Legacy data redistribution using direct ChunkStorage I/O
		void
		redistribute_data_legacy();

		/// New data redistribution using DataMigrationExecutor pipeline
		void
		redistribute_data_v2();

		static void
		expand_abt_v2(void* _arg);

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

		void
		shrink_start(int old_server_conf, int new_server_conf,

+31 −2

Original line number	Diff line number	Diff line
		@@ -25,11 +25,28 @@ namespace rpc {

		MigrationStatus
		DataMigrationExecutor::execute(std::vector<MigrationJob>& jobs,
		MigrationProgressCallback progress) {
		MigrationProgressCallback progress,
		JobHandler handler) {
		size_t total = jobs.size();
		size_t done = 0;
		size_t succeeded = 0;
		size_t failed = 0;

		if(handler) {
		for(size_t i = 0; i < jobs.size(); ++i) {
		int ret = handler(jobs[i]);
		if(ret == 0) {
		++succeeded;
		} else {
		++failed;
		}
		++done;

		if(progress) {
		progress(done, total);
		}
		}
		} else {
		for(size_t i = 0; i < jobs.size(); ++i) {
		success_count_.fetch_add(1);
		total_bytes_.fetch_add(4096);
		@@ -40,6 +57,7 @@ DataMigrationExecutor::execute(std::vector<MigrationJob>& jobs,
		progress(done, total);
		}
		}
		}

		if(succeeded == total)
		return MigrationStatus::Success;
		@@ -51,17 +69,28 @@ DataMigrationExecutor::execute(std::vector<MigrationJob>& jobs,
		MigrationStatus
		DataMigrationExecutor::execute_batched(std::vector<MigrationJob>& jobs,
		size_t batch_size,
		MigrationProgressCallback progress) {
		MigrationProgressCallback progress,
		JobHandler handler) {
		size_t total = jobs.size();
		size_t done = 0;
		size_t succeeded = 0;
		size_t failed = 0;

		for(size_t i = 0; i < total; i += batch_size) {
		size_t batch_end = std::min(i + batch_size, total);
		for(size_t j = i; j < batch_end; ++j) {
		if(handler) {
		int ret = handler(jobs[j]);
		if(ret == 0) {
		++succeeded;
		} else {
		++failed;
		}
		} else {
		success_count_.fetch_add(1);
		total_bytes_.fetch_add(4096);
		++succeeded;
		}
		++done;

		if(progress) {

+2 −2

Original line number	Diff line number	Diff line
		@@ -62,8 +62,8 @@ rpc_srv_expand_start(const tl::request& req,
		try {
		// if maintenance mode is already set, error is thrown -- not allowed
		GKFS_DATA->maintenance_mode(true);
		GKFS_DATA->malleable_manager()->expand_start(in.old_server_conf,
		in.new_server_conf);
		GKFS_DATA->malleable_manager()->expand_start(
		in.old_server_conf, in.new_server_conf, in.new_hosts_file);
		out.err = 0;
		} catch(const std::exception& e) {
		GKFS_DATA->spdlogger()->error("{}() Failed to start expansion: '{}' ",