Fix IO500 sfind wrapper validation and server-scaled CPU requests

################################################################################

NUM_NODES=10

GKFS_FIND_PROCESS=10

GKFS_SERVERS=$SLURM_JOB_NUM_NODES

GKFS_FIND=sfind

unset LD_PRELOAD

srun -N $NUM_NODES -n $GKFS_FIND_PROCESS --overlap --overcommit --mem=0 --oversubscribe --export=ALL env LD_PRELOAD=${GKFS} $GKFS_FIND "$@" -M $GKFS_MNT -S $GKFS_SERVERS --server-side -C

#!/bin/bash

# scripts/aggregate_sfind_results.sh

# Robustly aggregates gfind_results.rank-*.txt files

total_found=0

total_checked=0

# Enable nullglob so *.txt doesn't return literal string if no files match

shopt -s nullglob

files=(gfind_results.rank-*.txt)

if [ ${#files[@]} -eq 0 ]; then

    echo "No result files found (gfind_results.rank-*.txt)."

    exit 1

fi

echo "Found ${#files[@]} result files. Aggregating..."

for file in "${files[@]}"; do

    if [ ! -s "$file" ]; then

         echo "Warning: File $file is empty or missing. Skipping."

         continue

    fi

    # Read the line. Using -r to prevent backslash interpretation.

    if read -r line < "$file"; then

        # Expected format: MATCHED <found>/<checked>

        # Example: MATCHED 123/4567

        # Remove prefix "MATCHED " if present

        if [[ "$line" == MATCHED* ]]; then

            val_str="${line#MATCHED }"

        else

            # Try to handle cases where MATCHED might be missing or different

            val_str="$line" 

        fi

        # Split by '/'

        # found is everything before /

        found="${val_str%%/*}"

        # checked is everything after /

        checked="${val_str##*/}"

        # Validate that they are numbers

        if [[ "$found" =~ ^[0-9]+$ ]] && [[ "$checked" =~ ^[0-9]+$ ]]; then

            total_found=$((total_found + found))

            total_checked=$((total_checked + checked))

        else

            echo "Error: Invalid number format in $file: '$line' -> found='$found' checked='$checked'"

            # If set -e is active in parent script, we might want to exit? 

            # Or just warn and continue. Warn is safer for now.

        fi

    else

        echo "Warning: Could not read line from $file"

set -euo pipefail

# External IO500 find adapter. Keep all scan artifacts on shared storage.

fail() { echo "pfind.sh: $*" >&2; exit 1; }

: "${SLURM_JOB_NUM_NODES:?}" "${GKFS_MNT:?}" "${GKFS_HOSTS_FILE:?}"

intercept=$(readlink -f "${GKFS_INTERCEPT:-${GKFS:-}}")

[[ -r "$intercept" ]] || fail 'interceptor missing'

sfind=$(readlink -f "$(dirname "$(readlink -f "$0")")/sfind")

[[ -x "$sfind" ]] || fail "sfind missing: $sfind"

[[ -s "$GKFS_HOSTS_FILE" ]] || fail 'hostfile missing or empty'

export LIBGKFS_HOSTS_FILE="$GKFS_HOSTS_FILE"

# Count active entries, not calibration comments. Detect stale hostfiles.

servers=$(awk 'NF && $1 !~ /^#/ {n++} END {print n+0}' "$GKFS_HOSTS_FILE")

[[ "$servers" -eq "$SLURM_JOB_NUM_NODES" ]] || fail 'hostfile/allocation server count mismatch'

nodes=$((SLURM_JOB_NUM_NODES < 10 ? SLURM_JOB_NUM_NODES : 10))

# Match the largest server partition; sfind caps workers to its own partition.

cpus=$(((servers + nodes - 1) / nodes))

((cpus <= 20)) || cpus=20

# Resolve a relative -newer argument before switching working directory.

args=("$@")

for ((i=0; i<${#args[@]}; i++)); do

    if [[ "${args[i]}" == -newer ]]; then

        ((i+1 < ${#args[@]})) || fail 'missing -newer argument'

        args[i+1]=$(readlink -f "${args[i+1]}")

    fi

done

echo "MATCHED ${total_found}/${total_checked}"

exit 0

base="${CASE_OUT:-$(dirname "$GKFS_HOSTS_FILE")}/find-invocations"

# Never preload the interceptor into Slurm, Python aggregation, or the adapter.

unset LD_PRELOAD

mkdir -p "$base"

work=$(mktemp -d "$base/find.XXXXXXXX")

work=$(readlink -f "$work")

printf 'pfind.sh: artifacts=%s nodes=%s servers=%s cpus=%s\n' "$work" "$nodes" "$servers" "$cpus" >&2

printf '%s\n' "$sfind" "$intercept" "$GKFS_HOSTS_FILE" > "$work/paths.txt"

cd "$work"

set +e

srun --overlap --nodes="$nodes" --ntasks="$nodes" --ntasks-per-node=1 \

    --cpus-per-task="$cpus" --cpu-bind=none --kill-on-bad-exit=1 \

    --export=ALL --chdir="$work" \

    env LD_PRELOAD="$intercept" SFIND_NUM_THREADS="$cpus" \

    "$sfind" "${args[@]}" -M "$GKFS_MNT" -S "$servers" --server-side -C \

    > "$work/scan.stdout" 2> "$work/scan.stderr"

rc=$?

set -e

printf '%s\n' "$rc" > "$work/scan.exit"

if ((rc != 0)); then

    cat "$work/scan.stderr" >&2

    echo "pfind.sh: scan failed (exit $rc); artifacts: $work" >&2

    exit "$rc"

fi

/usr/bin/python3 - "$nodes" <<'PY'

import pathlib,re,sys

n=int(sys.argv[1]); found=checked=0

files=list(pathlib.Path('.').glob('gfind_results.rank-*.txt'))

expected={f'gfind_results.rank-{rank}.txt' for rank in range(n)}

if {p.name for p in files} != expected:

    sys.exit('pfind.sh: missing or unexpected rank result files')

for rank in range(n):

    p=pathlib.Path(f'gfind_results.rank-{rank}.txt')

    match=re.fullmatch(r'MATCHED ([0-9]+)/([0-9]+)\n',p.read_text())

    if not match:

        sys.exit(f'pfind.sh: malformed or empty result: {p}')

    f,c=map(int,match.groups())

    if f>c:

        sys.exit(f'pfind.sh: matched exceeds checked: {p}')

    found+=f; checked+=c

print(f'MATCHED {found}/{checked}',flush=True)

PY

"""Exercise the external IO500 adapter without Slurm or running daemons."""

import json

import os

from pathlib import Path

import shutil

import subprocess

import pytest

ROOT = Path(__file__).parents[2]

@pytest.mark.parametrize("servers,cpus", [(2, 1), (64, 7), (125, 13), (200, 20), (240, 20)])

@pytest.mark.parametrize("mode", ["ok", "fail", "missing", "empty", "bad", "inverted", "extra"])

def test_pfind_results_and_resources(tmp_path, servers, cpus, mode):

    binaries = tmp_path / "bin"

    binaries.mkdir()

    wrapper = binaries / "pfind.sh"

    shutil.copy(ROOT / "examples/gfind/pfind.sh", wrapper)

    wrapper.chmod(0o755)

    sfind = binaries / "sfind"

    sfind.write_text("#!/bin/sh\nexit 0\n")

    sfind.chmod(0o755)

    srun = binaries / "srun"

    srun.write_text('''#!/usr/bin/python3

import json, os, pathlib, sys

args = sys.argv[1:]

pathlib.Path('srun-args.json').write_text(json.dumps(args))

n = int(next(a.split('=', 1)[1] for a in args if a.startswith('--ntasks=')))

mode = os.environ['MOCK_MODE']

if mode == 'fail':

    sys.exit(7)

for rank in range(n):

    if mode == 'missing' and rank == n - 1:

        continue

    text = 'MATCHED 2/10\\n'

    if rank == 0:

        text = {'empty': '', 'bad': 'MATCHED nope\\n',

                'inverted': 'MATCHED 11/10\\n'}.get(mode, text)

    pathlib.Path(f'gfind_results.rank-{rank}.txt').write_text(text)

if mode == 'extra':

    pathlib.Path('gfind_results.rank-999.txt').write_text('MATCHED 2/10\\n')

print('rank diagnostics must not leak into the summary')

''')

    srun.chmod(0o755)

    interceptor = tmp_path / "intercept.so"

    interceptor.touch()

    hosts = tmp_path / "hosts"

    hosts.write_text("# calibration\n" + "".join(

        f"node{i} addr{i}\n" for i in range(servers)))

    output = tmp_path / "output"

    output.mkdir()

    # A stale result outside the invocation must never be included.

    (output / "gfind_results.rank-0.txt").write_text("MATCHED 999/999\n")

    timestamp = tmp_path / "timestamp"

    timestamp.touch()

    env = dict(os.environ, PATH=str(binaries) + ":" + os.environ["PATH"],

               GKFS_INTERCEPT=str(interceptor), GKFS_HOSTS_FILE=str(hosts),

               GKFS_MNT="/virtual", CASE_OUT=str(output),

               SLURM_JOB_NUM_NODES=str(servers), MOCK_MODE=mode)

    result = subprocess.run(

        [str(wrapper), "/virtual/data", "-newer", "timestamp", "-name", "*01*"],

        cwd=tmp_path, env=env, capture_output=True, text=True, timeout=30)

    assert (result.returncode == 0) == (mode == "ok"), result.stderr

    if mode == "ok":

        ranks = min(servers, 10)

        assert result.stdout == f"MATCHED {ranks * 2}/{ranks * 10}\n"

    else:

        assert result.stdout == ""

    if mode == "fail":

        assert result.returncode == 7

    invocation, = (output / "find-invocations").iterdir()

    args = json.loads((invocation / "srun-args.json").read_text())

    assert f"--cpus-per-task={cpus}" in args

    assert f"SFIND_NUM_THREADS={cpus}" in args

    assert f"--ntasks={min(servers, 10)}" in args

    assert str(timestamp) in args

 No newline at end of file