Baseline Pattern: Single-Node Service + Workers

One allocation claims one node and a bounded amount of CPU and time. These are the only lines that change between centers.

#!/usr/bin/env bash
#SBATCH --job-name=bssw-baseline
#SBATCH --nodes=1
#SBATCH --ntasks=1
#SBATCH --cpus-per-task=4
#SBATCH --time=00:10:00
#SBATCH --output=slurm-%j.out

set -Eeuo pipefail

WORKER_COUNT=${WORKER_COUNT:-2}
TASK_COUNT=${TASK_COUNT:-4}
READINESS_TIMEOUT=${READINESS_TIMEOUT:-30}
IMAGE_PATH=${IMAGE_PATH:-${WORKFLOW_ROOT}/container/baseline.sif}

The coordinator is launched inside the container on loopback. Nothing else starts until the readiness probe confirms it can actually serve — not just that the process exists.

# Launch the coordinator on loopback inside the allocation.
apptainer exec --cleanenv --bind "${RUNTIME_DIR}:/work" "${IMAGE_PATH}" \
  python3 "${APP_ROOT}/coordinator.py" \
  --input /work/input.json --results /work/results.json &

# Wait for it to publish its loopback endpoint.
while [[ ! -s "${RUNTIME_DIR}/endpoint.json" ]]; do
  sleep 0.05
done

# Run the semantic readiness probe (bounded HTTP /health contract).
ENDPOINT=$(cat "${RUNTIME_DIR}/endpoint.json" | python3 -c \
  'import json,sys; print(json.load(sys.stdin)["endpoint"])')
python3 "${APP_ROOT}/readiness.py" \
  --endpoint "${ENDPOINT}" --task-count "${TASK_COUNT}" \
  --timeout "${READINESS_TIMEOUT}"

Workers pull work from the coordinator over loopback. Only after every worker finishes does the verifier check the results — a completed process is not yet a successful result.

# Start one exclusive step per worker.
for ((i=1; i<=WORKER_COUNT; i++)); do
  printf -v worker_id 'worker-%02d' "$i"
  srun apptainer exec --bind "${RUNTIME_DIR}:/work" "${IMAGE_PATH}" \
    python3 "${APP_ROOT}/worker.py" \
    --endpoint "${ENDPOINT}" --worker-id "${worker_id}" &
done
wait

# Verify: schema, count, uniqueness, and expected content.
apptainer exec --bind "${RUNTIME_DIR}:/work" "${IMAGE_PATH}" \
  python3 "${APP_ROOT}/verify.py" \
  --input /work/input.json --results /work/results.json \
  --expected /work/expected.json --output /work/result.json

echo "BSSW workflow completed: result is in the runtime directory."

The probe polls the coordinator /health endpoint on loopback with a hard deadline. It passes only when the coordinator reports the expected schema and task count — so workers never start against a half-initialized service.

deadline = time.monotonic() + args.timeout
while time.monotonic() < deadline:
    try:
        with urllib.request.urlopen(f"{args.endpoint}/health") as response:
            health = json.load(response)
        if response.status == 200 and health == {
            "status": "ready",
            "schemaVersion": 1,
            "taskCount": args.task_count,
        }:
            raise SystemExit(0)   # ready
    except (OSError, ValueError, urllib.error.URLError):
        pass
    time.sleep(min(args.interval, deadline - time.monotonic()))
raise SystemExit(24)   # timed out

The /health response is the contract the readiness probe checks. Reporting taskCount means the coordinator has parsed its input and has work to hand out before any worker asks.

def health(self) -> dict:
    return {
        "status": "ready",
        "schemaVersion": 1,
        "taskCount": len(self.tasks),
    }

def claim(self) -> dict:
    # A worker asks for work; the coordinator hands out one task
    # or reports that all tasks are done (or that more may arrive).
    if self.pending:
        task_id = self.pending.popleft()
        return {"status": "task", "task": self.tasks[task_id]}
    status = "done" if len(self.results) == len(self.tasks) else "wait"
    return {"status": status}

The verifier checks that every task produced exactly one result, that results are unique, and that they match the expected outputs. Only then does it atomically write the success marker — the single artifact a completed run is judged by.

# Load tasks, actual results, and the expected results.
tasks = load_tasks(args.input)
actual = json.load(open(args.results))["results"]
expected = json.load(open(args.expected))["results"]

seen = set()
for result in actual:
    if result["taskId"] in seen:
        raise ValueError("duplicate task id")
    if result["taskId"] not in {e["taskId"] for e in expected}:
        raise ValueError("unexpected task id")
    seen.add(result["taskId"])

# Every task must be covered exactly once.
if seen != {task["id"] for task in tasks}:
    raise ValueError("results do not cover each task exactly once")

# Atomically write the success marker.
success = {
    "schemaVersion": 1,
    "status": "success",
    "taskCount": len(tasks),
    "results": sorted(actual, key=lambda r: r["taskId"]),
}
atomic_json(args.output, success)