Commit ceab36ad authored by Ramon Nou's avatar Ramon Nou
Browse files

Merge branch 'rnou/fix_expand' into 'master' 

Fix repeated Random Slicing expansion layout

Random slicing with cutshift has issues on multiple expand. This MR solves the issue.

See merge request !321
parents ca82063b aa627cf5

scripts/run/master-0.9.6

+6 −2
Changes for scripts/run/master-0.9.6: 6 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -71,8 +71,12 @@ setenv GKFS_BASE $BASE/gkfs-master 
# Useful defaults (can be overridden by user)

setenv        GKFS_MNT            /dev/shm/gkfs_mnt

setenv        GKFS_ROOT           $env(TMPDIR)/gkfs_root

setenv        GKFS_HOSTS_FILE     $env(HOME)/gkfs_hosts_$env(SLURM_JOB_ID).txt

setenv        LIBGKFS_HOSTS_FILE  $env(HOME)/gkfs_hosts_$env(SLURM_JOB_ID).txt

set job_id 1

if { [info exists env(SLURM_JOB_ID)] && $env(SLURM_JOB_ID) ne "" } {

    set job_id $env(SLURM_JOB_ID)

}

setenv        GKFS_HOSTS_FILE     $env(HOME)/gkfs_hosts_${job_id}.txt

setenv        LIBGKFS_HOSTS_FILE  $env(HOME)/gkfs_hosts_${job_id}.txt



prepend-path    LD_LIBRARY_PATH     $LIBS

prepend-path    PATH                $BASE/gkfs-master/bin

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: