Fix repeated Random Slicing expansion layout (8c614a62) · Commits · hpc / gekkofs

src/daemon/malleability/malleable_manager.cpp

+34 −7

Changes for src/daemon/malleability/malleable_manager.cpp: 34 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -1264,6 +1264,15 @@ MalleableManager::mutate_start(int old_server_conf, int new_server_conf,
		// after state.
		auto markers = gkfs::malleable::parse_hostfile_markers(new_hosts_file);
		gkfs::malleable::validate_hostfile_markers(markers);
		// Peers can replace the shared hostfile's interval comments concurrently.
		// The running distributor is the authoritative old placement for this
		// daemon, even after another peer publishes the new layout.
		const auto current_rs =
		std::dynamic_pointer_cast<gkfs::rpc::RandomSlicingDistributor>(
		RPC_DATA->distributor());
		const auto previous_rs_intervals = current_rs
		? current_rs->get_intervals()
		: vector<gkfs::rpc::Interval>{};
		vector<pair<string, string>> hosts;
		for(const auto& e : markers.active) {
		hosts.emplace_back(e.hostname, e.uri);
		@@ -1432,6 +1441,21 @@ MalleableManager::mutate_start(int old_server_conf, int new_server_conf,
		}
		}

		if(!previous_rs_intervals.empty() &&
		current_rs->hosts_size() == before_hosts.size()) {
		auto* old_rs = dynamic_cast<gkfs::rpc::RandomSlicingDistributor*>(
		old_distributor.get());
		auto* retained_old_rs =
		dynamic_cast<gkfs::rpc::RandomSlicingDistributor*>(
		old_distributor_.get());
		if(!old_rs \|\| !retained_old_rs \|\|
		!old_rs->set_intervals(previous_rs_intervals) \|\|
		!retained_old_rs->set_intervals(previous_rs_intervals)) {
		throw runtime_error(
		"Invalid old random-slicing intervals for mutate");
		}
		}

		unordered_map<string, bool> before_keys;
		for(const auto& host : before_hosts) {
		before_keys.emplace(host_key(host), true);
		@@ -1464,27 +1488,30 @@ MalleableManager::mutate_start(int old_server_conf, int new_server_conf,

		if(rs && cutshift && (!added_ids.empty() \|\| !removed_old_ids.empty()) &&
		!before_hosts.empty()) {
		auto comments =
		gkfs::malleable::parse_rs_interval_comments(new_hosts_file);
		vector<gkfs::rpc::Partition> updated_partitions;

		if(removed_old_ids.empty()) {
		auto old_intervals =
		comments.empty()
		previous_rs_intervals.empty()
		? vector<gkfs::rpc::Interval>{}
		: remap_existing_rs_comments(
		comments, before_hosts, hosts);
		to_interval_comments(
		previous_rs_intervals),
		before_hosts, hosts);
		updated_partitions =
		old_intervals.empty()
		? make_equal_rs_partitions_for_before_hosts(
		before_hosts, hosts)
		: intervals_to_partitions(old_intervals);
		} else {
		auto old_intervals = comments.empty()
		auto old_intervals =
		previous_rs_intervals.empty()
		? vector<gkfs::rpc::Interval>{}
		: rs_comments_to_old_id_intervals(
		comments, before_hosts);
		if(!comments.empty() && old_intervals.empty()) {
		to_interval_comments(
		previous_rs_intervals),
		before_hosts);
		if(!previous_rs_intervals.empty() && old_intervals.empty()) {
		GKFS_DATA->spdlogger()->warn(
		"{}() No usable pre-shrink RS interval comments found; rebuilding equal old layout",
		__func__);

tests/integration/malleability/test_mutate_repeated_expand.py

+85 −9

Changes for tests/integration/malleability/test_mutate_repeated_expand.py: 85 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -52,10 +52,14 @@ def _hostfile_entry_for(hostfile, daemon):

		def _write_workspace(hostfile, active, adding=None):
		adding = adding or []
		# CutShift records the current layout in comments. The next mutation must
		# start from that layout, not from a newly generated equal split.
		layout = [line + "\n" for line in Path(hostfile).read_text().splitlines()
		if line.startswith(("# GKFS_RS_INTERVAL ", "# GKFS_RS_LAYOUT "))]
		entries = [_hostfile_entry_for(hostfile, d) + "\n" for d in active]
		entries += ["+" + _hostfile_entry_for(hostfile, d) + "\n" for d in adding]
		with open(hostfile, "w") as hf:
		hf.writelines(entries)
		hf.writelines(entries + layout)


		def _run_mutate_cmd(gkfs_shell, hosts_file, args, timeout=120):
		@@ -146,21 +150,70 @@ def _snapshot_chunks(daemons, paths, label):
		print(f"CHUNK_SNAPSHOT {label} {json.dumps(snapshot, sort_keys=True)}", flush=True)


		def _try_read_md5(client, path):
		"""Return the file checksum, or None when stat/open/read fails."""
		ret = client.stat(path)
		if ret.retval != 0:
		return None
		size = ret.statbuf.st_size
		if client.open(path, os.O_RDONLY, 0).retval == -1:
		return None
		ret = client.read(path, size)
		if ret.retval != size or ret.buf is None:
		return None
		return hashlib.md5(ret.buf).hexdigest()


		def _chunk_census(daemons, path, expected_md5):
		"""Report chunk-0 copies and whether a full expected copy exists."""
		lines, found = [], False
		for daemon in daemons:
		chunk = Path(daemon.rootdir) / "chunks" / path.name / "0"
		if chunk.is_file():
		data = chunk.read_bytes()
		md5 = hashlib.md5(data).hexdigest()
		match = md5 == expected_md5
		found = found or match
		mark = " <- expected contents" if match else ""
		lines.append(f" {daemon.address}: {len(data)} bytes, md5 {md5}{mark}")
		else:
		lines.append(f" {daemon.address}: no chunk")
		return "\n".join(lines), found


		FILE_COUNT = 6
		ROUND_SIZES = [64 * 1024, 96 * 1024, 128 * 1024]
		ROUND_SIZES = [64 * 1024, 96 * 1024, 128 * 1024, 160 * 1024, 192 * 1024]


		@pytest.mark.parametrize("strategy", ["simple_hash", "random_slicing"])
		@pytest.mark.parametrize(
		"strategy,cutshift",
		[
		("simple_hash", False),
		("random_slicing", False),
		("random_slicing", True),
		],
		ids=["simple_hash", "random_slicing", "random_slicing_cutshift"],
		)
		def test_repeated_expand_keeps_the_latest_contents(
		strategy,
		cutshift,
		monkeypatch,
		gkfwd_daemon_factory,
		gkfs_shell,
		request,
		):
		monkeypatch.setenv("GKFS_DISTRIBUTION_STRATEGY", strategy)
		monkeypatch.setenv("GKFS_DAEMON_KEEP_HOSTS_FILE", "ON")
		settings = {
		"GKFS_DISTRIBUTION_STRATEGY": strategy,
		"GKFS_DAEMON_KEEP_HOSTS_FILE": "ON",
		"GKFS_RANDOM_SLICING_CUTSHIFT": "ON" if cutshift else "OFF",
		}
		for key, value in settings.items():
		monkeypatch.setenv(key, value)
		client = request.getfixturevalue("gkfs_client")
		# These fixtures can be instantiated before the test-local environment is
		# changed; keep the CLI and client on the same distribution settings.
		for env in (client._env, gkfs_shell._env):
		env.update(settings)

		daemons = [gkfwd_daemon_factory.create()]
		try:
		@@ -189,6 +242,10 @@ def test_repeated_expand_keeps_the_latest_contents(
		time.sleep(2)

		_write_workspace(hostfile, daemons[:-1], adding=[added])
		if cutshift and round_index > 0:
		assert "# GKFS_RS_INTERVAL " in hostfile.read_text(), (
		"Previous CutShift layout lost before the next expansion"
		)

		status = _run_mutate_cmd(gkfs_shell, hostfile, "mutate status")
		assert "No mutate running/finished." in status.stderr.decode()
		@@ -199,14 +256,33 @@ def test_repeated_expand_keeps_the_latest_contents(
		# The expansion really completed, so a failure below cannot be
		# explained by a redistribution still in flight.
		_assert_markers_consumed(hostfile)
		if cutshift:
		assert "# GKFS_RS_LAYOUT " in hostfile.read_text()
		_snapshot_chunks(daemons, paths, f"round-{round_index}-after-expand")

		for path in paths:
		got = _read_md5(client, path, daemons[0].logdir, hostfile)
		assert got == expected[path], (
		got = _try_read_md5(client, path)
		if got != expected[path]:
		census, found = _chunk_census(daemons, path, expected[path])
		if got is None and found:
		verdict = "Expected data is on disk, but stat or read failed."
		elif got is None:
		verdict = "Stat or read failed; no expected chunk-0 copy found."
		elif found:
		verdict = (
		"Expected data is on disk, but the client read other"
		" contents (check metadata size and data placement)."
		)
		else:
		verdict = (
		"The client read other contents; no expected chunk-0"
		" copy was found."
		)
		pytest.fail(
		f"round {round_index}: after expanding to {len(daemons)}"
		f" daemons, {path.name} returned contents that are not the"
		" ones written before this expansion"
		f" daemons, {path.name} read back {got or 'nothing'},"
		f" expected {expected[path]}.\n{verdict}\n"
		f"Where chunk 0 physically lives:\n{census}"
		)
		finally:
		for daemon in daemons: