Enable Random Slicing CutShift expansion with replicas (b8ddfb05) · Commits · hpc / gekkofs

src/common/rpc/random_slicing_distributor.cpp

+22 −2

Changes for src/common/rpc/random_slicing_distributor.cpp: 22 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -319,12 +319,32 @@ RandomSlicingDistributor::reconfigure() {
		bool
		RandomSlicingDistributor::set_intervals(
		const std::vector<Interval>& intervals) {
		if(!validate_intervals(intervals, hosts_size_)) {
		unsigned int interval_hosts = 0;
		std::set<host_t> interval_host_ids;
		for(const auto& interval : intervals) {
		interval_hosts = std::max(interval_hosts, interval.host_id + 1);
		interval_host_ids.insert(interval.host_id);
		}
		const auto effective_hosts = std::max(hosts_size_, interval_hosts);
		if(interval_host_ids.size() != effective_hosts) {
		return false;
		}
		for(unsigned int host = 0; host < effective_hosts; ++host) {
		if(interval_host_ids.find(host) == interval_host_ids.end()) {
		return false;
		}
		}
		if(!validate_intervals(intervals, effective_hosts)) {
		return false;
		}

		std::vector<Partition> partitions;
		partitions.reserve(hosts_size_);
		partitions.reserve(effective_hosts);
		all_hosts_.resize(effective_hosts);
		for(unsigned int host = 0; host < effective_hosts; ++host) {
		all_hosts_[host] = host;
		}
		hosts_size_ = effective_hosts;
		for(auto host : all_hosts_) {
		Partition partition;
		partition.host_id = host;

tests/integration/malleability/test_mutate_distributors_integrity.py

+66 −1

Changes for tests/integration/malleability/test_mutate_distributors_integrity.py: 66 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -33,7 +33,9 @@ FILE_SIZE = 64 * 1024



		def _run_mutate_cmd(gkfs_shell, hosts_file, args, timeout=120, logdir=None):
		def _run_mutate_cmd(
		gkfs_shell, hosts_file, args, timeout=120, logdir=None
		):
		cmd_str = f'LIBGKFS_HOSTS_FILE="{hosts_file}" gkfs_malleability {args}'
		cmd = gkfs_shell.script(cmd_str, intercept_shell=False, timeout=timeout)
		diagnostics = ""
		@@ -49,6 +51,7 @@ def _run_mutate_cmd(gkfs_shell, hosts_file, args, timeout=120, logdir=None):
		f"gkfs_malleability {args} failed\n"
		f"stdout: {cmd.stdout.decode()}\n"
		f"stderr: {cmd.stderr.decode()}"
		f"\nhostfile:\n{Path(hosts_file).read_text() if Path(hosts_file).exists() else '<missing>'}"
		f"{diagnostics}"
		)
		return cmd
		@@ -250,3 +253,65 @@ def test_mutate_add_remove_swap_keeps_data_and_chunks_distributed(
		finally:
		for daemon in daemons:
		daemon.shutdown()


		def test_random_slicing_cutshift_expand_with_replication(
		monkeypatch,
		gkfwd_daemon_factory,
		gkfs_client,
		gkfs_shell,
		request,
		):
		"""CutShift expansion preserves data and replica copies."""
		monkeypatch.setenv("GKFS_DISTRIBUTION_STRATEGY", "random_slicing")
		monkeypatch.setenv("GKFS_RANDOM_SLICING_CUTSHIFT", "ON")
		monkeypatch.setenv("GKFS_DAEMON_NUM_REPL", "1")
		monkeypatch.setenv("LIBGKFS_NUM_REPL", "1")

		# These fixtures are created before the test body. Update their client
		# environments explicitly after applying the test-local replication setup.
		for env in (gkfs_client._env, gkfs_shell._env):
		env.update(
		{
		"GKFS_DISTRIBUTION_STRATEGY": "random_slicing",
		"GKFS_RANDOM_SLICING_CUTSHIFT": "ON",
		"LIBGKFS_NUM_REPL": "1",
		}
		)

		daemons = []
		try:
		old_daemons = [
		gkfwd_daemon_factory.create(),
		gkfwd_daemon_factory.create(),
		]
		daemons.extend(old_daemons)
		time.sleep(3)

		md5_map = {}
		for index in range(3):
		path = old_daemons[0].mountdir / f"random_slicing_repl_{index}.dat"
		md5_map[path] = _write_file(gkfs_client, path, FILE_SIZE)
		_verify_md5_map(gkfs_client, md5_map)

		added = gkfwd_daemon_factory.create(expand_mode=True)
		daemons.append(added)
		time.sleep(2)

		hostfile = Path(old_daemons[0].hostfile)
		_write_workspace(hostfile, old_daemons, adding=[added])
		_run_mutate_cmd(
		gkfs_shell,
		hostfile,
		"mutate start",
		timeout=340,
		logdir=old_daemons[0].logdir,
		)
		_wait_for_mutate_done(gkfs_shell, hostfile)
		_run_mutate_cmd(gkfs_shell, hostfile, "mutate finalize")

		_verify_md5_map(gkfs_client, md5_map, old_daemons[0].logdir, hostfile)
		assert _chunk_files(*(daemon.rootdir for daemon in daemons))
		finally:
		for daemon in daemons:
		daemon.shutdown()
		No newline at end of file

tests/unit/test_random_slicing_distributor.cpp

+15 −0

Changes for tests/unit/test_random_slicing_distributor.cpp: 15 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -137,6 +137,21 @@ TEST_CASE("RandomSlicingDistributor keeps primary placement unchanged by replica
		}
		}

		TEST_CASE("RandomSlicingDistributor accepts intervals for added hosts",
		"[common][distributor][random_slicing][cutshift]") {
		auto d = gkfs::rpc::RandomSlicingDistributor(0, 2);
		const std::vector<gkfs::rpc::Interval> intervals = {
		{0.0f, 1.0f / 3.0f, 0},
		{1.0f / 3.0f, 2.0f / 3.0f, 1},
		{2.0f / 3.0f, 1.0f, 2},
		};

		REQUIRE(d.set_intervals(intervals));
		REQUIRE(d.hosts_size() == 3);
		REQUIRE(d.locate_data("/added", 0, 0) < 3);
		REQUIRE(d.locate_data("/added", 0, 1) < 3);
		}

		TEST_CASE("RandomSlicingDistributor bounds checking", "[common][distributor][random_slicing]") {
		auto d = gkfs::rpc::RandomSlicingDistributor(0, 10);