diff --git a/presto/scripts/summarize_coordinator_queries.py b/presto/scripts/summarize_coordinator_queries.py new file mode 100755 index 00000000..d8d026b8 --- /dev/null +++ b/presto/scripts/summarize_coordinator_queries.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Summarize coordinator query JSON files into a TSV report. + +Usage: + summarize_coordinator_queries.py + +Reads /queries/*.json (one file per query) and writes: + /summary.tsv — tab-separated summary of all queries + /operator_summaries/ — per-query operatorSummaries JSON (if present) +""" + +import json +import pathlib +import sys + +if len(sys.argv) < 2: + sys.exit("Usage: summarize_coordinator_queries.py ") + +out = pathlib.Path(sys.argv[1]) +summary_rows = [ + [ + "query_id", + "state", + "error_type", + "error_name", + "elapsed", + "execution", + "query_preview", + ] +] + +for path in sorted((out / "queries").glob("*.json")): + try: + with path.open() as f: + query = json.load(f) + except json.JSONDecodeError as e: + print(f"Warning: skipping {path.name}: {e}", file=sys.stderr) + continue + + stats = query.get("queryStats") or {} + error_code = query.get("errorCode") or {} + sql = (query.get("query") or "").replace("\n", " ") + if len(sql) > 180: + sql = sql[:177] + "..." + + summary_rows.append( + [ + query.get("queryId", path.stem), + query.get("state", ""), + query.get("errorType", ""), + error_code.get("name", ""), + stats.get("elapsedTime", ""), + stats.get("executionTime", ""), + sql, + ] + ) + + operators = stats.get("operatorSummaries") or [] + if operators: + operators_dir = out / "operator_summaries" + operators_dir.mkdir(exist_ok=True) + with (operators_dir / f"{query.get('queryId', path.stem)}.operators.json").open("w") as f: + json.dump(operators, f, indent=2) + +with (out / "summary.tsv").open("w") as f: + for row in summary_rows: + f.write("\t".join(str(value) for value in row) + "\n") diff --git a/presto/slurm/presto-nvl72/README.md b/presto/slurm/presto-nvl72/README.md index 39cdc553..286632df 100644 --- a/presto/slurm/presto-nvl72/README.md +++ b/presto/slurm/presto-nvl72/README.md @@ -105,18 +105,276 @@ This step only needs to repeat when the worker image's stat tracking changes ``` Submits a benchmark sbatch job, polls until completion, and prints a summary. -Results land under `result_dir/`. See `./launch-run.sh --help` for the full -flag list (queries filter, output path, GDS toggle, profiling, metrics, …). +Results land under an immutable `result_dir_/` directory. A later launch +does not delete or write into it, so it is safe to `scp` while another benchmark +runs. `latest_result_dir.txt` contains the newest job-specific directory name; +resolve that small pointer once before a copy rather than copying a stale legacy +`result_dir/` or following a mutable symlink. See `./launch-run.sh --help` for +the full flag list (queries filter, output path, GDS toggle, profiling, +metrics, …). **Prerequisites:** worker + coord images on disk; data from step 1; analyzed metastore from step 2 (either local or shared). +### CPU and GPU NUMA placement + +`CLUSTER_{CPU,GPU}_USE_NUMA` is a Boolean switch; `0` means unbound and does +not mean NUMA node 0. CPU jobs discover CPU-bearing DRAM nodes on every +allocated host and ignore accelerator/HBM-only NUMA IDs. Local worker slots are +balanced in contiguous groups. On the two-socket, four-GPU GB200 topology this +gives: + +| Workers per host (`-g`) | CPU worker mapping | +|---:|---| +| 1 | `0` | +| 2 | `0,1` | +| 4 | `0,0,1,1` | + +Each CPU worker is bound with both `--cpunodebind` and `--membind`; its driver +and memory defaults are sized from its share of that CPU/DRAM node rather than +from host `MemTotal`, which also includes the four HBM proximity domains. +Every selected host must report the same CPU topology. Each worker log records +`numactl --show` plus `Cpus_allowed_list` and `Mems_allowed_list` after entering +the requested policy and before `exec`-ing `presto_server`. + +GPU jobs retain their explicit `CLUSTER_GPU_NUMA_GPUS_PER_NODE` mapping. For +the topology above, setting it to `2` binds GPU workers 0–1 to CPU node 0 and +GPU workers 2–3 to CPU node 1. Values such as `2,10,18,26` in the +`nvidia-smi topo -m` **GPU NUMA ID** column are HBM nodes and are not valid CPU +binding targets. + +For a one-worker historical comparison only, `--legacy-cpu-numa` reproduces +the old inconsistent behavior: it uses the deprecated +`numactl --cpubind=0 --membind=0` spelling while retaining configuration sized +for the complete host. On this topology the process still has only CPUs 0–71; +`task.max-drivers-per-task=144` oversubscribes those 72 cores rather than +granting 144 cores. The option requires `--cpu -g 1`, emits a warning, and +records the effective policy in the worker log. It is not a supported scaling +configuration. + +Before startup, the job verifies that the coordinator, worker HTTP, and +exchange ports are free on each host. On exit it terminates only the exact +coordinator/worker `srun` clients it launched, waits for their steps, and +verifies port release. Listener ownership, command lines, and Slurm cgroups +are retained in the result logs when startup or cleanup fails. +Exchange ports are tested with an actual wildcard bind as well as `ss`, so a +closed UCX listener that is no longer in `LISTEN` but is still not reusable is +not mistaken for a free port. Preflight and teardown wait up to 90 seconds by +default; override `PRESTO_PORT_PREFLIGHT_TIMEOUT` or +`PRESTO_PORT_RELEASE_TIMEOUT` when needed. + +For a CPU UCX validation run, `--verify-cpu-ucx` waits up to 60 seconds for +every generated HTTP+3 exchange port to be claimed before queries begin. For a +multi-worker run, it then requires the retained query details to contain a +`UcxCpuRowExchange` or `UcxCpuRowPartitionedOutput` operator. A one-worker run +checks listener readiness but skips that runtime operator assertion because it +cannot prove inter-worker transport. The readiness check does not depend on UCX +log-message spelling or timing. The option is CLI-only: an inherited +`VERIFY_CPU_UCX` environment variable cannot enable it. Use an exchange-heavy +query with at least two workers for end-to-end transport verification; +`PRESTO_CPU_UCX_READY_TIMEOUT` changes the readiness timeout. + ### Override images for a single run ```bash ./launch-run.sh -n 2 -s 100 -w my-worker-image -c my-coord-image ``` +### Query-scoped CPU flamegraph + +`--perf` attaches the compute host's kernel-matched `perf` to worker 0. It does +not require `perf` inside the Enroot image. Keep this separate from `-p/--profile` +(nsys), and select one representative query and iteration: + +```bash +./launch-run.sh --cpu -n 10 -g 1 -s 1000 -q 18 -i 2 \ + --perf --profile-iterations 1 \ + -w -c +``` + +The sidecar preserves or raises its inherited host `nofile` limit to the hard +limit granted to the Slurm job, then tests the requested event against the live +worker. For each capture, it snapshots every live worker TID and partitions +them among bounded `perf record` processes. Every recorder starts with events +disabled. Recorders are initialized serially and must answer a control `ping` +before the query-side enable barrier; a transient initialization failure is +retried up to three times. The query starts only after every recorder +acknowledges that every target event is open. Inheritance covers worker threads +created after that snapshot. An unrecoverable attach-time descriptor or setup +failure therefore aborts before the measured query starts. Missing host `perf`, +a low hard descriptor limit, +`perf_event_paranoid`, PMU, ptrace, or Slurm cgroup restrictions likewise fail +the run explicitly instead of producing an empty profile. Capture metadata +records the live worker thread/mapping counts, shard membership, and each +recorder's open descriptor count for capacity diagnosis. Every active raw +capture is retained on node-local +`${SLURM_TMPDIR:-/tmp}` (`PERF_CAPTURE_TMPDIR` overrides the staging base). +Stopping a query waits only until every recorder has stopped, closed its output, +and queued those node-local files. This prevents home/NFS publication and +symbolization from blocking pytest or perturbing the next measured query. + +After the complete query suite, the sidecar atomically publishes all queued +`perf.data` shards into the result directory before worker teardown. It also +snapshots the worker's executable mappings into a portable `symfs/`; this keeps +symbolization independent of Enroot's private mount namespace and permits +reprocessing after the job. Publication progress and byte counts are written to +`logs/perf_worker_0.log`. By default, expensive derived-report generation is +deferred: + +```text +result_dir_/profiles/perf/worker_0/ +├── symfs/ # exact mapped executables/shared libraries +├── symbol-manifest.tsv +└── Q18_iter1/ + ├── perf.data.part000 + ├── perf.data.part001 + ├── perf-shards.tsv + ├── perf-startup-attempts.tsv + ├── perf-target-tids.txt + ├── perf-attached-tids.part000.txt + ├── perf-dropped-tids.tsv + └── capture-metadata.txt +``` + +One-shard captures retain the conventional `perf.data` name. For a multi-shard +capture, `process-perf-captures.sh` symbolizes each raw file and concatenates +the sample streams into one `perf.script.gz` and one aggregate flamegraph. The +text report labels each raw shard because its percentages are shard-local. +The generated `flamegraph.html` is self-contained: click a frame to zoom into +that subtree, use **Reset Zoom** to return, and use the search box to highlight +matching functions. + +The recorder snapshots disjoint worker TID sets, but worker-pool threads can +exit while the shards are being initialized. Before every startup attempt, it +filters each shard against `/proc//task`; an exited TID is recorded in +`perf-dropped-tids.tsv` instead of making every retry fail with `ESRCH`. The +per-shard `perf-attached-tids.*.txt` files preserve the exact final target set. +Inheritance remains enabled for threads created after attachment. + +Generate `perf-report.txt`, `perf.script.gz`, `stacks.folded`, and +`flamegraph.html` later from the login/head node with: + +```bash +./process-perf-captures.sh result_dir_ + +# Or process one capture: +./process-perf-captures.sh result_dir_ --query Q18_iter1 + +# Omit inline-function expansion for a much faster, still symbolized graph: +./process-perf-captures.sh result_dir_ \ + --query Q18_iter1 --no-inline +``` + +An already generated aggregate `stacks.folded` is architecture-independent. +To upgrade or regenerate only its interactive HTML on the head node—without +rerunning `perf script` or requesting a compute allocation—run: + +```bash +capture=result_dir_JOBID/profiles/perf/worker_0/Q18_iter1 +python3 ../../scripts/perf_flamegraph.py \ + --input-folded "${capture}/stacks.folded" \ + --output "${capture}/flamegraph.html" \ + --title "Presto worker 0: Q18_iter1" +``` + +When called outside Slurm, the script uses the CPU partition/account from +`~/.cluster_config.env` and starts one blocking `srun` compute allocation. No +host name or rail is selected or hard-coded. It requests one CPU per raw +recorder shard, copies the capture and `symfs/` to +`${SLURM_TMPDIR:-/tmp}`, symbolizes the shards in parallel, renders the combined +stream, and atomically copies only the derived artifacts back. This ensures an +Arm64 Grace capture is unwound by an Arm64 `perf`, avoids repeated random I/O +against the shared result directory, and prevents copy permission warnings by +not preserving shared-filesystem metadata in the temporary tree. `--jobs N` +caps shard concurrency, `--time HH:MM:SS` changes the processing allocation, +and `--local` bypasses `srun` when the command is already running on a +compatible compute host. + +`--no-inline` is passed to both `perf report` and `perf script`, including +across the automatic `srun` relaunch. It disables inline-function expansion, +not DWARF stack unwinding: physical call chains, symbols, raw data, and the +portable `symfs/` remain intact. The chosen mode, compute hostname, +architecture, shard count, and parallelism are recorded in +`postprocess-metadata.txt`. Re-running with a different inline mode regenerates +the derived files automatically; `--force` is only required to repeat the same +mode. + +Pass `--perf-postprocess` to the original launch only when those derived files +must be generated inside the allocation. That work still waits until all +queries and raw-file publication have completed, so it cannot bias later query +captures. The postprocessor uses the bundled dependency-free Python renderer +over the combined `perf script` output, so it needs neither a network download +nor the old Perl FlameGraph scripts. + +The bundled renderer rejects output when fewer than 5% of sampled call chains +contain a known symbol, rather than calling an all-`[unknown]` graph successful. +It also refuses any raw directory whose metadata lacks the complete +ready/enable/stop/publish sequence, whose shard manifest contains a failed +recorder, or whose published shard count is incomplete. Diagnostic partial +logs and metadata from a failed attach are retained, while its unusable +pre-query partial `perf.data` files are discarded instead of being copied to +shared storage. A failed attach therefore cannot silently become a folded stack +file or flamegraph. +If rendering fails, the command exits nonzero and publishes +`flamegraph-unavailable.txt`, `perf.script.gz`, the text report, and diagnostics; +the raw `perf.data` shards and matching `symfs/` are never removed or modified. + +Profiled benchmark runs stop after the first failed query, preventing a broken +sampler setup from wasting the remainder of a full suite. + +Use `--perf-frequency`, `--perf-event`, `--perf-call-graph`, +`--perf-user-regs`, `--perf-nofile-limit`, and +`--perf-threads-per-shard` to override the defaults (`19`, `cpu-clock:u`, +`dwarf,8192`, `auto`, a `65536` minimum, and `128` target TIDs per recorder). +The sidecar never lowers a larger inherited soft limit and raises it to the +job's hard limit when permitted. `perf record` gets 120 seconds to flush and +stop after each query; `--perf-record-stop-timeout` overrides that watchdog. +These conservative +sampling defaults are intentional on a 144-core worker: higher-frequency DWARF +capture can lose samples, create hundreds of megabytes per short query, and +materially perturb the query being measured. Profile-run timings should not be +posted as benchmark timings. If `-o/--output-path` is given, the profile is +included in the normal result-directory copy. + +`cpu-clock:u` is intentionally a software, user-space, on-CPU sampling event. +It produces the CPU-time flamegraph needed for hotspot analysis without relying +on a particular hardware counter. On this Grace/NVIDIA Linux 6.14 stack, perf +can still expand an event across the PMU CPU map for every target thread. A +warmed Presto worker has over one thousand threads, so one recorder can exceed +the job's 131072-descriptor hard limit. The TID shards bound descriptors per +recorder without dropping threads or restricting the CPUs on which they may be +sampled. `--perf-event cycles:u` remains available for hardware-cycle sampling +and uses the same sharding mechanism. + +On an Arm64 host that advertises SVE, perf's automatic DWARF register set also +contains the extended `vg` pseudo-register. Linux's software clock PMUs reject +that extended register even though the hardware PMU accepts it. The default +`--perf-user-regs auto` therefore supplies every base Arm64 GPR while omitting +`vg` for `cpu-clock`/`task-clock`; explicit register lists are passed through, +and `perf-default` restores perf's own selection. Preflight uses a short real +`perf record`, not `perf stat`, so unsupported sampling/register combinations +fail before the benchmark begins. + +Artifact publication waits for completion by default and is ultimately bounded +by the Slurm job time limit, because prematurely killing a copy can lose the +only node-local raw capture. `--perf-finalize-timeout ` supplies an +explicit nonzero bound when desired. + +For copy-back automation, resolve the pointer once: + +```bash +run_dir="$(cat latest_result_dir.txt)" +rsync -aP "${run_dir}/" /destination/"${run_dir}/" + +# For a quick diagnostic copy that intentionally omits large perf artifacts: +rsync -aP --exclude '/profiles/perf/' "${run_dir}/" /destination/"${run_dir}/" +``` + +By default, `--perf` also fails preflight when the worker executable contains +neither `.symtab` nor `.debug_info`, since an address-only flamegraph is not a +useful result. Set `PERF_REQUIRE_SYMBOLS=0` only when intentionally collecting +raw addresses for later symbolization with a matching external symbol build. + --- ## Entry points (reference) @@ -227,6 +485,10 @@ presto-nvl72/ ├── functions.sh # Coordinator/worker/queries helpers ├── echo_helpers.sh # Logging helpers ├── profiler_functions.sh # nsys profiling helpers (bind-mounted into workers) +├── perf_profiler_functions.sh # query-side perf token handshake +├── perf-worker-sampler.sh # host-side worker-0 capture orchestration +├── perf-record-sharded.sh # bounded TID-shard perf record processes +├── process-perf-captures.sh # deferred perf report/flamegraph generation │ ├── # Slurm wrappers + inner execution scripts ├── gen-tpch-data.slurm # Step 1 sbatch entry @@ -237,7 +499,7 @@ presto-nvl72/ │ ├── # Runtime output ├── logs/ # Coord/worker/cli logs (created per run) -└── result_dir/ # Benchmark results (created per run) +└── result_dir_/ # Immutable benchmark results per Slurm job ``` --- diff --git a/presto/slurm/presto-nvl72/collect-coordinator-queries.sh b/presto/slurm/presto-nvl72/collect-coordinator-queries.sh new file mode 100755 index 00000000..f25a244b --- /dev/null +++ b/presto/slurm/presto-nvl72/collect-coordinator-queries.sh @@ -0,0 +1,159 @@ +#!/usr/bin/env bash +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Collect retained Presto query JSON from a live Slurm benchmark coordinator. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "${SCRIPT_DIR}" + +source ./defaults.env + +PORT="${PORT:-${CLUSTER_DEFAULT_PORT:-9200}}" +OUT_DIR="${OUT_DIR:-}" +JOB_ID="" +SERVER="" + +usage() { + cat < [-o output_dir] [-p port] + $0 --server http://host:port [-o output_dir] + +Fetches the coordinator's retained /v1/query list and one full +/v1/query/ JSON file per query into output_dir. + +Defaults: + output_dir: result_dir_/coordinator_queries (job ID mode) + coordinator_queries (--server mode) + port: ${PORT} +EOF +} + +while [[ $# -gt 0 ]]; do + case "$1" in + -o|--output-dir) + [[ -n "${2:-}" ]] || { echo "Error: $1 requires a value" >&2; exit 2; } + OUT_DIR="$2" + shift 2 + ;; + -p|--port) + [[ -n "${2:-}" ]] || { echo "Error: $1 requires a value" >&2; exit 2; } + PORT="$2" + shift 2 + ;; + --server) + [[ -n "${2:-}" ]] || { echo "Error: $1 requires a value" >&2; exit 2; } + SERVER="${2%/}" + shift 2 + ;; + -h|--help) + usage + exit 0 + ;; + -*) + echo "Unknown option: $1" >&2 + usage >&2 + exit 2 + ;; + *) + if [[ -z "${JOB_ID}" ]]; then + JOB_ID="$1" + else + echo "Unexpected argument: $1" >&2 + usage >&2 + exit 2 + fi + shift + ;; + esac +done + +if [[ -z "${JOB_ID}" && -z "${SERVER}" ]]; then + usage >&2 + exit 2 +fi + +# Default output directory incorporates the job ID when available so each run +# lands in the same result_dir_/ tree as the benchmark output. +if [[ -z "${OUT_DIR}" ]]; then + if [[ -n "${JOB_ID}" ]]; then + OUT_DIR="result_dir_${JOB_ID}/coordinator_queries" + else + OUT_DIR="coordinator_queries" + fi +fi + +if [[ -n "${JOB_ID}" ]]; then + NODELIST="$(squeue -j "${JOB_ID}" -h -o '%N' | awk 'NR==1{print $1}')" + if [[ -z "${NODELIST}" || "${NODELIST}" == "(null)" ]]; then + echo "Could not find allocated nodes for job ${JOB_ID}. Is it still running?" >&2 + exit 1 + fi + + COORD="$(scontrol show hostnames "${NODELIST}" | head -1)" + if [[ -z "${COORD}" ]]; then + echo "Could not resolve coordinator node from node list: ${NODELIST}" >&2 + exit 1 + fi + + SERVER="http://${COORD}:${PORT}" +fi + +mkdir -p "${OUT_DIR}/queries" + +[[ -n "${JOB_ID}" ]] && echo "Job: ${JOB_ID}" +[[ -n "${COORD:-}" ]] && echo "Coordinator: ${COORD}" +echo "Server: ${SERVER}" +echo "Output: ${OUT_DIR}" + +fetch_from_coord() { + local path="$1" + if [[ -n "${JOB_ID}" ]]; then + srun --jobid "${JOB_ID}" -w "${COORD}" --ntasks=1 --overlap \ + bash -lc "curl -fsS --max-time 30 '${SERVER}${path}'" + else + curl -fsS --max-time 30 "${SERVER}${path}" + fi +} + +echo "Fetching coordinator query list..." +fetch_from_coord "/v1/query" > "${OUT_DIR}/queries.json" +fetch_from_coord "/v1/info" > "${OUT_DIR}/info.json" || true +fetch_from_coord "/v1/node" > "${OUT_DIR}/nodes.json" || true + +mapfile -t QUERY_IDS < <( + python3 -c ' +import json, sys +with open(sys.argv[1]) as f: + data = json.load(f) +for q in data: + qid = q.get("queryId") + if qid: + print(qid) +' "${OUT_DIR}/queries.json" +) + +if [[ "${#QUERY_IDS[@]}" -eq 0 ]]; then + echo "No retained queries found in ${OUT_DIR}/queries.json" + exit 0 +fi + +echo "Fetching ${#QUERY_IDS[@]} full query JSON file(s)..." +for query_id in "${QUERY_IDS[@]}"; do + if fetch_from_coord "/v1/query/${query_id}" > "${OUT_DIR}/queries/${query_id}.json"; then + echo " ${query_id}" + else + echo " ${query_id} (failed to fetch)" >&2 + rm -f "${OUT_DIR}/queries/${query_id}.json" + fi +done + +python3 "${SCRIPT_DIR}/../../scripts/summarize_coordinator_queries.py" "${OUT_DIR}" + +echo "Wrote:" +echo " ${OUT_DIR}/queries.json" +echo " ${OUT_DIR}/queries/*.json" +echo " ${OUT_DIR}/operator_summaries/*.operators.json" +echo " ${OUT_DIR}/summary.tsv" diff --git a/presto/slurm/presto-nvl72/functions.sh b/presto/slurm/presto-nvl72/functions.sh index 86808ddd..07048a2b 100644 --- a/presto/slurm/presto-nvl72/functions.sh +++ b/presto/slurm/presto-nvl72/functions.sh @@ -2,6 +2,20 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +PRESTO_SCRIPT_FUNCTIONS_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../scripts" && pwd)" +source "${PRESTO_SCRIPT_FUNCTIONS_DIR}/common_functions.sh" + +# Track only the srun clients started by this batch shell. Cleanup never uses a +# broad pkill, so it cannot terminate another user's process or an unrelated +# process in the allocation. +COORD_SRUN_PID="" +declare -a WORKER_SRUN_PIDS=() +declare -a WORKER_SRUN_IDS=() +declare -a WORKER_SRUN_NODES=() +declare -A CPU_NUMA_NODE_BY_WORKER_SLOT=() +declare -A CLUSTER_PORT_SPECS_BY_NODE=() +CLUSTER_STOPPED=0 + # Echo the ",:" bind-mount fragment for the host's # miniforge3 install if it exists, else nothing. Used by every container # that needs to run Python via init_python_virtual_env / run_py_script.sh: @@ -80,6 +94,38 @@ function generate_configs { echo "-XX:-UseContainerSupport" >> ${CONFIGS}/etc_common/jvm.config } +function resolve_node_ipv4_for_interface { + [ $# -ne 2 ] && echo_error "$0 expected arguments 'node' and 'interface'" + local node=$1 iface=$2 + srun -N1 -w "${node}" --ntasks=1 --overlap \ + /bin/bash -lc "ip -o -4 addr show dev '${iface}' | awk '{print \$4}' | cut -d/ -f1 | head -n 1" +} + +# Resolve an RDMA device (for example mlx5_0:1) to its Linux netdev and IPv4 +# address on a particular allocated node. This is deliberately performed on +# the compute node: NVL72 odd/even trays attach to different leaf groups, and +# interface names need not be globally encoded in the cluster configuration. +function resolve_node_ipv4_for_ucx_device { + [ $# -ne 2 ] && echo_error "$0 expected arguments 'node' and 'ucx_device'" + local node=$1 rdma_device=${2%%:*} + [[ "${rdma_device}" =~ ^[[:alnum:]_.-]+$ ]] || return 1 + + srun -N1 -w "${node}" --ntasks=1 --overlap \ + /bin/bash -lc ' + rdma_device=$1 + net_dir="/sys/class/infiniband/${rdma_device}/device/net" + [[ -d "${net_dir}" ]] || exit 1 + shopt -s nullglob + net_paths=("${net_dir}"/*) + ((${#net_paths[@]} > 0)) || exit 1 + iface="${net_paths[0]##*/}" + address="$(ip -o -4 addr show dev "${iface}" scope global \ + | sed -n "1{s/.* inet \\([^/ ]*\\).*/\\1/p;}")" + [[ -n "${address}" ]] || exit 1 + printf "%s %s\\n" "${iface}" "${address}" + ' _ "${rdma_device}" +} + # Takes a list of environment variables. Checks that each one is set and of non-zero length. function validate_environment_preconditions { local missing=() @@ -137,6 +183,8 @@ ${CONFIGS}/etc_coordinator/catalog/hive.properties:/opt/presto-server/etc/catalo ${DATA}:/var/lib/presto/data/hive/data/user_data,\ ${VT_ROOT}/.hive_metastore:/var/lib/presto/data/hive/metastore${extra_mounts} \ -- bash -lc "unset JAVA_HOME; export JAVA_HOME=/usr/lib/jvm/jre-17-openjdk; export PATH=/usr/lib/jvm/jre-17-openjdk/bin:\$PATH; ${script}" >> ${LOGS}/${log_file} 2>&1 & + COORD_SRUN_PID=$! + echo "Coordinator srun client PID: ${COORD_SRUN_PID}" else srun -w $COORD --ntasks=1 --overlap \ --container-remap-root \ @@ -206,17 +254,38 @@ run_coord_image "$COORD_SCRIPT" "coord" function run_worker { : "${ENABLE_GDS:=0}" : "${ENABLE_NSYS:=0}" - : "${NSYS_WORKER_ID:=0}" + : "${ENABLE_PERF:=0}" + : "${PERF_WORKER_ID:=0}" + # Migrate legacy NSYS_WORKER_ID (single int) → NSYS_WORKER_IDS (csv). + : "${NSYS_WORKER_IDS:=${NSYS_WORKER_ID:-0}}" - [ $# -ne 4 ] && echo_error "$0 expected arguments 'gpu_id', 'image', 'node_id', and 'worker_id'" + [ $# -ne 4 ] && echo_error "$0 expected arguments 'local_worker_id', 'image', 'node_id', and 'worker_id'" validate_environment_preconditions LOGS CONFIGS VT_ROOT COORD CUDF_LIB DATA - local gpu_id=$1 image=$2 node=$3 worker_id=$4 - # Pair each worker with the CPU NUMA domain its GPU is attached to. - # CLUSTER_NUMA_GPUS_PER_NODE controls how many GPUs share each NUMA domain - # (e.g., 2 means GPUs 0-1 on NUMA 0, GPUs 2-3 on NUMA 1). - local numa_node=$((gpu_id / ${CLUSTER_NUMA_GPUS_PER_NODE:-1})) - echo "running worker ${worker_id} with image ${image} on node ${node} with gpu_id ${gpu_id} numa_node ${numa_node}" + # The first argument is the worker's local slot on its host. For GPU jobs + # that slot is also the CUDA device id; for CPU jobs it is only an index + # into node-local interface/device lists. It restarts at zero on each host. + local local_worker_id=$1 image=$2 node=$3 worker_id=$4 + local gpu_id="${local_worker_id}" + local numa_node="" + if [[ "${VARIANT_TYPE}" == "gpu" ]]; then + # Pair each worker with the CPU NUMA domain its GPU is attached to. + # On GB200, GPUs 0-1 have CPU affinity 0-71 (node 0) and GPUs 2-3 + # have CPU affinity 72-143 (node 1). The GPU NUMA IDs reported by + # nvidia-smi are HBM nodes and are intentionally not used here. + numa_node=$((gpu_id / ${CLUSTER_NUMA_GPUS_PER_NODE:-1})) + elif [[ "${USE_NUMA}" == "1" ]]; then + numa_node="${CPU_NUMA_NODE_BY_WORKER_SLOT["${node}:${local_worker_id}"]:-}" + [[ -n "${numa_node}" ]] || \ + echo_error "Worker ${worker_id}: CPU NUMA placement was not prepared for ${node} local slot ${local_worker_id}" + fi + if [[ "${VARIANT_TYPE}" == "gpu" ]]; then + echo "running worker ${worker_id} with image ${image} on node ${node} with gpu_id ${gpu_id} numa_node ${numa_node}" + elif [[ "${USE_NUMA}" == "1" ]]; then + echo "running worker ${worker_id} with image ${image} on node ${node} with local_slot ${local_worker_id} cpu_numa_node ${numa_node}" + else + echo "running worker ${worker_id} with image ${image} on node ${node} with local_slot ${local_worker_id} (NUMA unbound)" + fi local worker_image worker_image=$(resolve_image_path "${image}") @@ -228,9 +297,93 @@ function run_worker { local worker_hive="${CONFIGS}/etc_worker_${worker_id}/catalog/hive.properties" local worker_data="${SCRIPT_DIR}/worker_data_${worker_id}" + # CPU UCX peers derive the destination listener port from the advertised + # Prestissimo HTTP task URL as HTTP+3. Derive the local listener from the + # same generated config rather than putting a single static port in the + # shared worker.env file; this stays correct for every worker ID. + local cpu_ucx_port="" + if [[ "${VARIANT_TYPE}" == "cpu" ]]; then + local worker_http_port + worker_http_port="$(awk -F= ' + /^[[:space:]]*http-server\.http\.port[[:space:]]*=/ { + gsub(/[[:space:]]/, "", $2) + print $2 + exit + } + ' "${worker_config}")" + if [[ ! "${worker_http_port}" =~ ^[0-9]+$ ]] || (( worker_http_port < 1 || worker_http_port > 65532 )); then + echo_error "Worker ${worker_id}: invalid or missing http-server.http.port in ${worker_config}" + fi + cpu_ucx_port=$((worker_http_port + 3)) + echo " Worker ${worker_id} CPU UCX listener: HTTP ${worker_http_port} + 3 = ${cpu_ucx_port}" + fi + # Each worker needs to be told how to access the coordianator sed -i "s+discovery\.uri.*+discovery\.uri=http://${COORD}:${PORT}+g" ${worker_config} + # Resolve a worker-specific UCX device from its local slot before entering + # the container. The same list is reused independently on every allocated + # node, so scheduling remains node-agnostic. Preserve the legacy per-GPU + # list and the generic UCX_NET_DEVICES multi-rail value as fallbacks. + local worker_ucx_net_device="${UCX_NET_DEVICES:-${CLUSTER_UCX_NET_DEVICES:-}}" + local ucx_net_devices_by_local_worker="${CLUSTER_UCX_NET_DEVICES_BY_LOCAL_WORKER:-}" + local ucx_net_devices_by_gpu="${CLUSTER_UCX_NET_DEVICES_BY_GPU:-}" + if [[ -n "${ucx_net_devices_by_local_worker}" ]]; then + IFS=',' read -ra ucx_net_device_entries <<< "${ucx_net_devices_by_local_worker}" + worker_ucx_net_device="${ucx_net_device_entries[${local_worker_id}]:-}" + [[ -n "${worker_ucx_net_device}" ]] || \ + echo_error "CLUSTER_UCX_NET_DEVICES_BY_LOCAL_WORKER has no entry for local worker ${local_worker_id}" + elif [[ -n "${ucx_net_devices_by_gpu}" ]]; then + IFS=',' read -ra ucx_net_device_entries <<< "${ucx_net_devices_by_gpu}" + worker_ucx_net_device="${ucx_net_device_entries[${gpu_id}]:-}" + [[ -n "${worker_ucx_net_device}" ]] || \ + echo_error "CLUSTER_UCX_NET_DEVICES_BY_GPU has no entry for GPU ${gpu_id}" + fi + [[ -n "${worker_ucx_net_device}" ]] && \ + echo " Worker ${worker_id} UCX device: ${worker_ucx_net_device}" + + # Keep an explicit interface override when supplied. Otherwise CPU UCX + # derives the advertised address from the first selected RDMA device on + # each actual host. For a multi-rail list this first device is also the + # listener/bootstrap rail; UCX remains free to use the full device list for + # data movement. The legacy per-GPU interface list remains available to GPU + # runs and to callers that intentionally want fixed local-slot pairing. + local internal_address_interface="${PRESTO_INTERNAL_ADDRESS_INTERFACE:-${CLUSTER_INTERNAL_ADDRESS_INTERFACE:-}}" + local internal_address_interface_by_gpu="${PRESTO_INTERNAL_ADDRESS_INTERFACE_BY_GPU:-${CLUSTER_INTERNAL_ADDRESS_INTERFACE_BY_GPU:-}}" + if [[ -n "${internal_address_interface_by_gpu}" ]]; then + local -a internal_address_interfaces + IFS=',' read -ra internal_address_interfaces <<< "${internal_address_interface_by_gpu}" + internal_address_interface="${internal_address_interfaces[${gpu_id}]:-}" + [[ -n "${internal_address_interface}" ]] || \ + echo_error "PRESTO_INTERNAL_ADDRESS_INTERFACE_BY_GPU/CLUSTER_INTERNAL_ADDRESS_INTERFACE_BY_GPU has no entry for GPU ${gpu_id}" + fi + + local internal_address="" address_source="" + if [[ -n "${internal_address_interface}" ]]; then + internal_address="$(resolve_node_ipv4_for_interface "${node}" "${internal_address_interface}")" + [[ -n "${internal_address}" ]] || \ + echo_error "Could not resolve ${internal_address_interface} IPv4 address on ${node}" + address_source="interface ${internal_address_interface}" + elif [[ "${VARIANT_TYPE}" == "cpu" && "${worker_ucx_net_device}" == mlx5_* ]]; then + local bootstrap_ucx_device="${worker_ucx_net_device%%,*}" + local discovered_address + discovered_address="$(resolve_node_ipv4_for_ucx_device "${node}" "${bootstrap_ucx_device}")" || \ + echo_error "Worker ${worker_id}: could not map ${bootstrap_ucx_device} to an IPv4 netdev on ${node}" + read -r internal_address_interface internal_address <<< "${discovered_address}" + [[ -n "${internal_address_interface}" && -n "${internal_address}" ]] || \ + echo_error "Worker ${worker_id}: incomplete RDMA address discovery for ${bootstrap_ucx_device} on ${node}" + address_source="auto ${bootstrap_ucx_device} -> ${internal_address_interface}" + fi + + if [[ -n "${internal_address}" ]]; then + if grep -q "^node\.internal-address=" "${worker_node}"; then + sed -i "s+^node\.internal-address=.*+node.internal-address=${internal_address}+g" "${worker_node}" + else + echo "node.internal-address=${internal_address}" >> "${worker_node}" + fi + echo " Worker ${worker_id} internal address: ${internal_address} (${address_source})" + fi + # Create unique data dir per worker. mkdir -p ${worker_data} mkdir -p ${worker_data}/hive/data/user_data @@ -239,6 +392,14 @@ function run_worker { local vt_worker_env_file="/var/worker_env_file" local vt_cufile_log_dir="/var/log/cufile" local vt_cufile_log="${vt_cufile_log_dir}/cufile_worker_${worker_id}.log" + local container_script_dir="${SCRIPT_DIR/${VT_ROOT}//workspace}" + [[ "${container_script_dir}" == /workspace/* ]] || \ + echo_error "SCRIPT_DIR must be below VT_ROOT (${VT_ROOT}) so the NUMA launcher is visible in the worker container: ${SCRIPT_DIR}" + local vt_numa_launcher="${container_script_dir}/numa-worker-launch.sh" + if [[ "${USE_NUMA:-0}" == "1" ]]; then + [[ -f "${SCRIPT_DIR}/numa-worker-launch.sh" ]] || \ + echo_error "NUMA worker launcher not found: ${SCRIPT_DIR}/numa-worker-launch.sh" + fi # GDS (GPU Direct Storage) is a GPU-only feature. On CPU variant, skip # the cufile/nvidia-fs bind-mounts even when ENABLE_GDS=1 — /etc/cufile.json @@ -257,12 +418,19 @@ function run_worker { local nsys_bin="" local nsys_launch_opts="" local vt_nsys_report_dir="/var/log/nsys" - if [[ "${ENABLE_NSYS}" == "1" && "${worker_id}" == "${NSYS_WORKER_ID}" ]]; then + local publish_perf_pid=0 + if [[ "${ENABLE_NSYS}" == "1" && ",${NSYS_WORKER_IDS}," == *",${worker_id},"* ]]; then # nsys must be on PATH inside the worker container. The recommended approach is # a symlink in the image build: # ln -sf /opt/nvidia/nsight-systems-cli//bin/nsys /usr/local/bin/nsys nsys_bin="nsys" - nsys_launch_opts="-t nvtx,cuda" + # nsys then uses its own built-in default trace set. + nsys_launch_opts="${NSYS_LAUNCH_OPTS:-}" + fi + if [[ "${ENABLE_PERF}" == "1" && "${worker_id}" == "${PERF_WORKER_ID}" ]]; then + publish_perf_pid=1 + # Remove a PID from a prior failed job before the new container starts. + rm -f "${LOGS}/.presto_worker_pid_w${worker_id}" fi # To re-enable verbose GLOG logging, add these flags to the srun call below @@ -297,13 +465,26 @@ function run_worker { fi # NVIDIA_VISIBLE_DEVICES triggers enroot's 98-nvidia.sh hook, which requires - # nvidia-container-cli on the host. CPU-only nodes typically don't have it, - # so the hook fails container start. Skip the export on the CPU variant. + # nvidia-container-cli on the host. CPU-only nodes typically do not have it, + # so this export remains GPU-only. local nvidia_env="" if [[ "${VARIANT_TYPE}" == "gpu" ]]; then nvidia_env=",NVIDIA_VISIBLE_DEVICES=all,NVIDIA_DRIVER_CAPABILITIES=compute,utility" fi + # Enroot's Mellanox hook exposes /dev/infiniband and the matching providers. + # GPU jobs default to the historical "all" behavior. Selecting an mlx5 + # device is also an explicit CPU opt-in: expose the matching RDMA devices + # and providers without requiring a second, redundant topology setting. + # Ordinary CPU-only hosts using UCX auto/TCP still avoid the hook. + if [[ -n "${CLUSTER_MELLANOX_VISIBLE_DEVICES:-}" ]]; then + export MELLANOX_VISIBLE_DEVICES="${CLUSTER_MELLANOX_VISIBLE_DEVICES}" + elif [[ "${VARIANT_TYPE}" == "gpu" || "${worker_ucx_net_device}" == *mlx5_* ]]; then + export MELLANOX_VISIBLE_DEVICES=all + else + unset MELLANOX_VISIBLE_DEVICES + fi + # Notes on nsys profiling # # In the worker container here, we use `nsys launch` to start the presto_server, @@ -333,6 +514,11 @@ function run_worker { # The token files live in /var/log/nsys, the same host directory bind-mounted # into both the worker and cli containers. + # Pyxis/Enroot share the host network and IPC namespaces unless explicitly + # asked to unshare them (and do not create a separate PID namespace here). + # That already matches Compose's network/ipc/pid host settings required by + # UCX same-host transports; do not add --container-unshare=ipc to this step. + UCX_NET_DEVICES="${worker_ucx_net_device}" \ srun -N1 -w $node --ntasks=1 --overlap \ --container-image=${worker_image} \ --container-remap-root \ @@ -349,12 +535,31 @@ ${WORKER_ENV_FILE}:${vt_worker_env_file},\ ${LOGS}:${vt_cufile_log_dir},\ ${LOGS}:${vt_nsys_report_dir}${driver_mounts}${gds_mounts:+,${gds_mounts}}${worker_extra_mounts} \ -- /bin/bash -c " -export LD_LIBRARY_PATH='${CUDF_LIB}':/usr/local/lib:\${LD_LIBRARY_PATH:-} - set -a source ${vt_worker_env_file} set +a +if [[ \"\${PRESTO_WORKER_UCX_LOCAL_LIB_FIRST:-0}\" == \"1\" ]]; then + export LD_LIBRARY_PATH=/usr/local/lib:'${CUDF_LIB}':\${LD_LIBRARY_PATH:-} +else + export LD_LIBRARY_PATH='${CUDF_LIB}':/usr/local/lib:\${LD_LIBRARY_PATH:-} +fi + +if [[ '${VARIANT_TYPE}' == 'cpu' ]]; then + if [[ -n \"\${VELOX_UCX_CPU_PORT:-}\" && \"\${VELOX_UCX_CPU_PORT}\" != '${cpu_ucx_port}' ]]; then + echo \"Worker ${worker_id}: overriding VELOX_UCX_CPU_PORT=\${VELOX_UCX_CPU_PORT} with generated HTTP+3 port ${cpu_ucx_port}\" + fi + export VELOX_UCX_CPU_EXCHANGE=\"\${VELOX_UCX_CPU_EXCHANGE:-1}\" + export VELOX_UCX_CPU_PORT='${cpu_ucx_port}' + if [[ \"\${VERIFY_CPU_UCX:-0}\" == '1' && \"\${VELOX_UCX_CPU_EXCHANGE}\" != '1' ]]; then + echo \"Worker ${worker_id}: --verify-cpu-ucx requires VELOX_UCX_CPU_EXCHANGE=1\" >&2 + exit 2 + fi + echo \"Worker ${worker_id}: CPU UCX enabled=\${VELOX_UCX_CPU_EXCHANGE}, port=\${VELOX_UCX_CPU_PORT}\" + echo \"Worker ${worker_id}: UCX_TLS=\${UCX_TLS:-unset}, UCX_NET_DEVICES=\${UCX_NET_DEVICES:-unset}\" + echo \"Worker ${worker_id}: UCX_MODULE_DIR=\${UCX_MODULE_DIR:-unset}\" +fi + if [[ '${VARIANT_TYPE}' == 'gpu' ]]; then export CUDA_VISIBLE_DEVICES=${gpu_id}; fi if [[ '${ENABLE_GDS}' == '1' ]]; then @@ -366,13 +571,29 @@ else fi echo \"Worker ${worker_id}: CUDA_VISIBLE_DEVICES=\${CUDA_VISIBLE_DEVICES:-none}, NUMA_NODE=${numa_node}\" +echo \"Worker ${worker_id}: UCX_NET_DEVICES=\${UCX_NET_DEVICES:-unset}\" echo \"Worker ${worker_id}: ENABLE_GDS=\${ENABLE_GDS:-unset}\" echo \"Worker ${worker_id}: ENABLE_NSYS=\${ENABLE_NSYS:-unset}\" +echo \"Worker ${worker_id}: ENABLE_PERF=\${ENABLE_PERF:-unset}\" echo \"Worker ${worker_id}: KVIKIO_COMPAT_MODE=\${KVIKIO_COMPAT_MODE:-unset}\" echo \"Worker ${worker_id}: CUFILE_LOGFILE_PATH=\${CUFILE_LOGFILE_PATH:-unset}\" echo \"Worker ${worker_id}: KVIKIO_TASK_SIZE=\${KVIKIO_TASK_SIZE:-unset}\" echo \"Worker ${worker_id}: KVIKIO_NTHREADS=\${KVIKIO_NTHREADS:-unset}\" +if [[ '${VARIANT_TYPE}' == 'cpu' && \"\${VERIFY_CPU_UCX:-0}\" == '1' ]]; then + echo \"Worker ${worker_id}: LD_LIBRARY_PATH=\${LD_LIBRARY_PATH:-unset}\" + if command -v ucx_info >/dev/null 2>&1; then + echo \"Worker ${worker_id}: UCX runtime version\" + ucx_info -v + else + echo \"Worker ${worker_id}: ucx_info not found\" >&2 + fi + if command -v ldd >/dev/null 2>&1; then + echo \"Worker ${worker_id}: presto_server UCX linkage\" + ldd /usr/bin/presto_server | grep -E 'libuc[mpxt]|not found' || true + fi +fi + if [[ -n '${nsys_bin}' ]]; then ( echo \"Worker ${worker_id}: nsys subshell started\" @@ -407,45 +628,669 @@ if [[ -n '${nsys_bin}' ]]; then echo \"Worker ${worker_id}: Nsight System program at ${nsys_bin}\" echo \"Worker ${worker_id}: running nsys launch\" if [[ '${USE_NUMA}' == '1' ]]; then - ${nsys_bin} launch ${nsys_launch_opts} numactl --cpubind=${numa_node} --membind=${numa_node} /usr/bin/presto_server --etc-dir=/opt/presto-server/etc + ${nsys_bin} launch ${nsys_launch_opts} /bin/bash ${vt_numa_launcher} ${numa_node} /usr/bin/presto_server --etc-dir=/opt/presto-server/etc else ${nsys_bin} launch ${nsys_launch_opts} /usr/bin/presto_server --etc-dir=/opt/presto-server/etc fi echo \"Worker ${worker_id}: nsys launch exited with code: \$?\" else - if [[ '${USE_NUMA}' == '1' ]]; then - numactl --cpubind=${numa_node} --membind=${numa_node} /usr/bin/presto_server --etc-dir=/opt/presto-server/etc + if [[ '${publish_perf_pid}' == '1' ]]; then + perf_pid_file='${vt_nsys_report_dir}/.presto_worker_pid_w${worker_id}' + rm -f \"\${perf_pid_file}\" + if [[ '${USE_NUMA}' == '1' ]]; then + /bin/bash ${vt_numa_launcher} ${numa_node} /usr/bin/presto_server --etc-dir=/opt/presto-server/etc & + else + /usr/bin/presto_server --etc-dir=/opt/presto-server/etc & + fi + server_pid=\$! + printf '%s\n' \"\${server_pid}\" > \"\${perf_pid_file}.tmp\" + mv \"\${perf_pid_file}.tmp\" \"\${perf_pid_file}\" + echo \"Worker ${worker_id}: published presto_server host PID \${server_pid} for perf\" + wait \"\${server_pid}\" + server_rc=\$? + rm -f \"\${perf_pid_file}\" + exit \"\${server_rc}\" + elif [[ '${USE_NUMA}' == '1' ]]; then + /bin/bash ${vt_numa_launcher} ${numa_node} /usr/bin/presto_server --etc-dir=/opt/presto-server/etc else /usr/bin/presto_server --etc-dir=/opt/presto-server/etc fi fi " > ${LOGS}/worker_${worker_id}.log 2>&1 & + WORKER_SRUN_PIDS+=("$!") + WORKER_SRUN_IDS+=("${worker_id}") + WORKER_SRUN_NODES+=("${node}") + echo " Worker ${worker_id} srun client PID: ${WORKER_SRUN_PIDS[$((${#WORKER_SRUN_PIDS[@]} - 1))]}" +} + +# ---------------------------------------------------------------------------- +# Host perf sidecar +# ---------------------------------------------------------------------------- + +# PID of the background srun that hosts perf-worker-sampler.sh. This is kept in +# the batch shell so success and failure paths can both finalize captures before +# Slurm tears down the allocation. +PERF_SIDECAR_SRUN_PID="" + +function start_perf_sampler { + [[ "${ENABLE_PERF:-0}" == "1" ]] || return 0 + + local worker_id="${PERF_WORKER_ID:-0}" + local sampler="${SCRIPT_DIR}/perf-worker-sampler.sh" + local sharded_recorder="${SCRIPT_DIR}/perf-record-sharded.sh" + local postprocess_script="${SCRIPT_DIR}/process-perf-captures.sh" + local result_dir="${RESULT_DIR:-${SCRIPT_DIR}/result_dir}" + local output_dir="${result_dir}/profiles/perf" + local ready_file="${LOGS}/.perf_sidecar_ready_w${worker_id}" + local error_file="${LOGS}/.perf_sidecar_error_w${worker_id}" + local done_file="${LOGS}/.perf_sidecar_done_w${worker_id}" + local sampler_log="${LOGS}/perf_worker_${worker_id}.log" + + [[ "${worker_id}" == "0" ]] || echo_error "This perf path currently supports worker 0 only" + [[ -f "${sampler}" ]] || echo_error "Perf sampler not found: ${sampler}" + [[ -f "${sharded_recorder}" ]] || echo_error "Sharded perf recorder not found: ${sharded_recorder}" + [[ -f "${postprocess_script}" ]] || echo_error "Perf postprocessor not found: ${postprocess_script}" + + mkdir -p "${output_dir}" + rm -f \ + "${ready_file}" \ + "${error_file}" \ + "${done_file}" \ + "${LOGS}/.perf_sidecar_shutdown_w${worker_id}" \ + "${LOGS}"/.perf_start_w"${worker_id}"_* \ + "${LOGS}"/.perf_started_w"${worker_id}"_* \ + "${LOGS}"/.perf_stop_w"${worker_id}"_* \ + "${LOGS}"/.perf_finished_w"${worker_id}"_* \ + "${LOGS}"/.perf_error_w"${worker_id}"_* + + echo "Starting host perf sidecar for worker ${worker_id} on ${COORD}..." + srun -N1 -w "${COORD}" --ntasks=1 --overlap --export=ALL \ + bash "${sampler}" \ + --worker-id "${worker_id}" \ + --control-dir "${LOGS}" \ + --output-dir "${output_dir}" \ + --postprocess-script "${postprocess_script}" \ + --postprocess "${PERF_POSTPROCESS:-0}" \ + --event "${PERF_EVENT:-cpu-clock:u}" \ + --frequency "${PERF_FREQUENCY:-19}" \ + --call-graph "${PERF_CALL_GRAPH:-dwarf,8192}" \ + --user-regs "${PERF_USER_REGS:-auto}" \ + --nofile-limit "${PERF_NOFILE_LIMIT:-65536}" \ + --threads-per-shard "${PERF_THREADS_PER_SHARD:-128}" \ + --record-stop-timeout "${PERF_RECORD_STOP_TIMEOUT:-120}" \ + >"${sampler_log}" 2>&1 & + PERF_SIDECAR_SRUN_PID=$! + + local waited + for waited in {1..180}; do + if [[ -f "${ready_file}" ]]; then + echo "Host perf sidecar is ready for worker ${worker_id}." + return 0 + fi + if [[ -s "${error_file}" ]]; then + cat "${error_file}" >&2 + echo_error "Host perf preflight failed; see ${sampler_log}" + fi + if ! kill -0 "${PERF_SIDECAR_SRUN_PID}" 2>/dev/null; then + wait "${PERF_SIDECAR_SRUN_PID}" || true + PERF_SIDECAR_SRUN_PID="" + echo_error "Host perf sidecar exited before becoming ready; see ${sampler_log}" + fi + sleep 1 + done + echo_error "Timed out waiting for host perf sidecar; see ${sampler_log}" +} + +function stop_perf_sampler { + [[ "${ENABLE_PERF:-0}" == "1" ]] || return 0 + + local worker_id="${PERF_WORKER_ID:-0}" + local shutdown_file="${LOGS}/.perf_sidecar_shutdown_w${worker_id}" + local done_file="${LOGS}/.perf_sidecar_done_w${worker_id}" + local error_file="${LOGS}/.perf_sidecar_error_w${worker_id}" + local sampler_log="${LOGS}/perf_worker_${worker_id}.log" + local finalize_timeout="${PERF_FINALIZE_TIMEOUT:-0}" + local sidecar_rc=0 + local timeout_rc=0 + + if [[ -z "${PERF_SIDECAR_SRUN_PID}" && -f "${done_file}" ]]; then + if [[ -s "${error_file}" ]]; then + cat "${error_file}" >&2 + return 1 + fi + return 0 + fi + [[ "${finalize_timeout}" =~ ^[0-9]+$ ]] || { + echo "Invalid PERF_FINALIZE_TIMEOUT=${finalize_timeout}; expected a non-negative integer" >&2 + return 1 + } + + touch "${shutdown_file}" + + if [[ -n "${PERF_SIDECAR_SRUN_PID}" ]]; then + if (( finalize_timeout > 0 )); then + if command -v timeout >/dev/null 2>&1; then + timeout --foreground "${finalize_timeout}s" bash -c ' + while kill -0 "$1" 2>/dev/null; do + sleep 1 + done + ' bash "${PERF_SIDECAR_SRUN_PID}" || timeout_rc=$? + else + local deadline=$((SECONDS + finalize_timeout)) + while kill -0 "${PERF_SIDECAR_SRUN_PID}" 2>/dev/null && (( SECONDS < deadline )); do + sleep 1 + done + if kill -0 "${PERF_SIDECAR_SRUN_PID}" 2>/dev/null; then + timeout_rc=124 + fi + fi + if (( timeout_rc == 124 )); then + echo "Timed out after ${finalize_timeout}s waiting for perf artifact publication; terminating its srun. See ${sampler_log}" >&2 + kill -TERM "${PERF_SIDECAR_SRUN_PID}" 2>/dev/null || true + elif (( timeout_rc != 0 )); then + echo "Error while waiting for perf artifact publication (status ${timeout_rc}); see ${sampler_log}" >&2 + kill -TERM "${PERF_SIDECAR_SRUN_PID}" 2>/dev/null || true + fi + fi + wait "${PERF_SIDECAR_SRUN_PID}" || sidecar_rc=$? + PERF_SIDECAR_SRUN_PID="" + fi + if [[ ! -f "${done_file}" ]]; then + echo "Perf sidecar did not publish its completion token; see ${sampler_log}" >&2 + return 1 + fi + if [[ -s "${error_file}" ]]; then + cat "${error_file}" >&2 + return 1 + fi + if (( sidecar_rc != 0 )); then + echo "Perf sidecar srun exited with status ${sidecar_rc}; see ${sampler_log}" >&2 + return "${sidecar_rc}" + fi + echo "Host perf raw captures published. Artifacts: ${RESULT_DIR:-${SCRIPT_DIR}/result_dir}/profiles/perf/worker_${worker_id}" } # ---------------------------------------------------------------------------- # Cluster lifecycle # ---------------------------------------------------------------------------- + +# Return success only while a tracked child exists and is not a zombie waiting +# for this shell to reap it. +function tracked_process_is_running { + local pid="${1:-}" state="" + [[ "${pid}" =~ ^[1-9][0-9]*$ ]] || return 1 + kill -0 "${pid}" 2>/dev/null || return 1 + state="$(ps -o stat= -p "${pid}" 2>/dev/null | tr -d '[:space:]')" + [[ -n "${state}" && "${state}" != Z* ]] +} + +# Discover the CPU-bearing NUMA nodes independently on every selected host. +# The generated worker configs are homogeneous, so reject a heterogeneous +# allocation instead of silently applying resource limits tuned for another +# node's topology. +function prepare_cpu_numa_layout { + CPU_NUMA_NODE_BY_WORKER_SLOT=() + [[ "${VARIANT_TYPE:-gpu}" == "cpu" ]] || return 0 + + if [[ "${USE_NUMA:-0}" != "1" ]]; then + echo "CPU NUMA binding is disabled; workers may use every CPU and allowed memory node on their host." + return 0 + fi + local common_functions="${PRESTO_SCRIPT_FUNCTIONS_DIR}/common_functions.sh" + local reference_signature="" node output line + for node in $(scontrol show hostnames "${SLURM_JOB_NODELIST}"); do + if ! output="$( + srun -N1 -w "${node}" --ntasks=1 --overlap \ + /bin/bash -s -- "${common_functions}" "${NUM_GPUS_PER_NODE}" <<'EOS' +set -euo pipefail +source "$1" +workers_per_host="$2" +discover_cpu_numa_topology + +cpu_ids="$(IFS=,; printf '%s' "${CPU_NUMA_NODE_IDS[*]}")" +cpu_counts="$(IFS=,; printf '%s' "${CPU_NUMA_NODE_CPU_COUNTS[*]}")" +cpu_memory="$(IFS=,; printf '%s' "${CPU_NUMA_NODE_MEMORY_GB[*]}")" +memory_ids="$(IFS=,; printf '%s' "${MEMORY_NUMA_NODE_IDS[*]}")" +printf 'TOPOLOGY|%s|%s|%s|%s|%s\n' \ + "${VISIBLE_NUMA_NODE_COUNT}" "${cpu_ids}" "${cpu_counts}" \ + "${cpu_memory}" "${memory_ids}" + +for ((slot = 0; slot < workers_per_host; ++slot)); do + printf 'MAP|%s|%s\n' \ + "${slot}" "$(cpu_numa_node_for_worker "${slot}" "${workers_per_host}")" +done +EOS + )"; then + echo "${output}" >&2 + echo_error "Could not discover CPU NUMA topology on ${node}" + fi + + local signature="" visible="" cpu_ids="" cpu_counts="" cpu_memory="" memory_ids="" + local slot="" cpu_node="" map_count=0 mapping="" + while IFS= read -r line; do + case "${line}" in + TOPOLOGY\|*) + IFS='|' read -r _ visible cpu_ids cpu_counts cpu_memory memory_ids <<< "${line}" + signature="${visible}|${cpu_ids}|${cpu_counts}|${cpu_memory}|${memory_ids}" + ;; + MAP\|*) + IFS='|' read -r _ slot cpu_node <<< "${line}" + if [[ ! "${slot}" =~ ^[0-9]+$ || ! "${cpu_node}" =~ ^[0-9]+$ ]]; then + echo_error "Invalid CPU NUMA mapping returned by ${node}: ${line}" + fi + CPU_NUMA_NODE_BY_WORKER_SLOT["${node}:${slot}"]="${cpu_node}" + mapping+="${mapping:+,}${slot}->${cpu_node}" + map_count=$((map_count + 1)) + ;; + "") + ;; + *) + echo "CPU NUMA discovery output from ${node}: ${line}" + ;; + esac + done <<< "${output}" + + [[ -n "${signature}" ]] || echo_error "CPU NUMA discovery on ${node} returned no topology record" + (( map_count == NUM_GPUS_PER_NODE )) || \ + echo_error "CPU NUMA discovery on ${node} returned ${map_count} worker mappings; expected ${NUM_GPUS_PER_NODE}" + + if [[ -z "${reference_signature}" ]]; then + reference_signature="${signature}" + elif [[ "${signature}" != "${reference_signature}" ]]; then + echo_error "CPU NUMA topology on ${node} differs from the coordinator host; heterogeneous CPU resource tuning is not supported" + fi + + echo "CPU NUMA topology on ${node}: visible=${visible}, CPU+DRAM nodes=${cpu_ids} (CPUs=${cpu_counts}, DRAM_GB=${cpu_memory}), all memory nodes=${memory_ids}" + echo "CPU NUMA worker mapping on ${node}: ${mapping}" + done +} + +# Inspect fixed coordinator, worker HTTP, and HTTP+3 exchange ports on one +# compute node. require-free waits until every port is reusable; require-bound +# waits until every selected port has been claimed. Exchange ports get an +# actual wildcard bind probe in addition to ss: UCX may own the sockaddr +# through a non-TCP listener, and a recently closed UCX listener can have no +# LISTEN entry while kernel socket state still makes a new bind fail with +# EADDRINUSE. The report includes Slurm cgroups, command lines, and matching +# TCP socket state where the kernel permits ownership inspection. +function run_port_inspection { + [ $# -ne 5 ] && echo_error "$0 expected arguments 'node', 'specs', 'mode', 'output_file', and 'wait_seconds'" + local node=$1 specs=$2 mode=$3 output_file=$4 wait_seconds=$5 + srun -N1 -w "${node}" --ntasks=1 --overlap \ + /bin/bash -s -- "${node}" "${specs}" "${mode}" "${wait_seconds}" \ + >"${output_file}" 2>&1 <<'EOS' +set -uo pipefail +node="$1" +specs="$2" +mode="$3" +wait_seconds="$4" + +case "${mode}" in + report|require-free|require-bound) + ;; + *) + echo "Port inspection on ${node}: invalid mode ${mode}" >&2 + exit 2 + ;; +esac +if ! command -v ss >/dev/null 2>&1; then + echo "Port inspection on ${node}: ss is not installed" >&2 + exit 2 +fi +if ! command -v python3 >/dev/null 2>&1; then + echo "Port inspection on ${node}: python3 is required for exchange-port bind probes" >&2 + exit 2 +fi + +probe_bind() { + python3 - "$1" <<'PY' +import socket +import sys + +port = int(sys.argv[1]) +try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("0.0.0.0", port)) +except OSError as error: + print( + f"bind(0.0.0.0:{port}) failed: " + f"[errno {error.errno}] {error.strerror}") + raise SystemExit(1) +PY +} + +inspect_once() { + local mismatch=0 entry role port matches pids pid bind_error connections state show_details + IFS=',' read -ra entries <<< "${specs}" + for entry in "${entries[@]}"; do + role="${entry%%:*}" + port="${entry##*:}" + state="free" + bind_error="" + connections="" + show_details=0 + matches="$(ss -H -ltnp 2>/dev/null \ + | awk -v wanted="${port}" '{ + count = split($4, fields, ":") + if (fields[count] == wanted) print + }')" + if [[ -n "${matches}" ]]; then + state="listen" + elif [[ "${role}" == *-exchange ]] && ! bind_error="$(probe_bind "${port}" 2>&1)"; then + state="bind-blocked" + connections="$(ss -H -tanp 2>/dev/null \ + | awk -v wanted="${port}" '{ + count = split($4, fields, ":") + if (fields[count] == wanted) print + }')" + fi + + case "${mode}" in + report) + case "${state}" in + listen) + echo "BUSY ${node} ${role} port ${port}" + show_details=1 + ;; + bind-blocked) + echo "BIND-BLOCKED ${node} ${role} port ${port}" + [[ -n "${bind_error}" ]] && echo "${bind_error}" + [[ -n "${connections}" ]] && echo "${connections}" + ;; + free) + echo "FREE ${node} ${role} port ${port}" + ;; + esac + ;; + require-free) + if [[ "${state}" == "listen" ]]; then + mismatch=1 + echo "BUSY ${node} ${role} port ${port}" + show_details=1 + elif [[ "${state}" == "bind-blocked" ]]; then + mismatch=1 + echo "BIND-BLOCKED ${node} ${role} port ${port}" + [[ -n "${bind_error}" ]] && echo "${bind_error}" + [[ -n "${connections}" ]] && echo "${connections}" + fi + ;; + require-bound) + if [[ "${state}" == "free" ]]; then + mismatch=1 + echo "MISSING ${node} ${role} port ${port}" + else + echo "READY ${node} ${role} port ${port} (${state})" + fi + ;; + esac + + if (( show_details == 1 )); then + echo "${matches}" + if command -v fuser >/dev/null 2>&1; then + pids="$(fuser -n tcp "${port}" 2>/dev/null || true)" + for pid in ${pids}; do + [[ "${pid}" =~ ^[0-9]+$ ]] || continue + ps -p "${pid}" -o pid=,ppid=,user=,stat=,lstart=,args= 2>/dev/null || true + if [[ -r "/proc/${pid}/cgroup" ]]; then + echo " cgroup:" + sed 's/^/ /' "/proc/${pid}/cgroup" + fi + if [[ -r "/proc/${pid}/cmdline" ]]; then + echo -n " cmdline: " + tr '\0' ' ' < "/proc/${pid}/cmdline" + echo + fi + done + fi + fi + done + return "${mismatch}" +} + +deadline=$((SECONDS + wait_seconds)) +while true; do + report="$(inspect_once)" + status=$? + if (( status == 0 )); then + [[ -n "${report}" ]] && echo "${report}" + exit 0 + fi + if [[ "${mode}" == "report" ]] || (( SECONDS >= deadline )); then + echo "${report}" + exit "${status}" + fi + sleep 1 +done +EOS +} + +function preflight_cluster_ports { + CLUSTER_PORT_SPECS_BY_NODE=() + local preflight_timeout="${PRESTO_PORT_PREFLIGHT_TIMEOUT:-${PRESTO_PORT_RELEASE_TIMEOUT:-90}}" + [[ "${preflight_timeout}" =~ ^[0-9]+$ ]] || \ + echo_error "Invalid PRESTO_PORT_PREFLIGHT_TIMEOUT=${preflight_timeout}" + + local worker_id=0 node local_worker_id config http_port exchange_port configured_exchange_port specs log_file + for node in $(scontrol show hostnames "${SLURM_JOB_NODELIST}"); do + specs="" + if [[ "${node}" == "${COORD}" ]]; then + specs="coordinator:${PORT}" + fi + for ((local_worker_id = 0; local_worker_id < NUM_GPUS_PER_NODE; ++local_worker_id)); do + config="${CONFIGS}/etc_worker_${worker_id}/config_native.properties" + http_port="$(awk -F= '/^http-server\.http\.port=/ {gsub(/[[:space:]]/, "", $2); print $2; exit}' "${config}")" + [[ "${http_port}" =~ ^[1-9][0-9]*$ ]] || \ + echo_error "Worker ${worker_id}: invalid HTTP port in ${config}" + (( http_port <= 65532 )) || \ + echo_error "Worker ${worker_id}: HTTP port ${http_port} leaves no valid HTTP+3 exchange port" + exchange_port=$((http_port + 3)) + configured_exchange_port="$(awk -F= '/^cudf\.exchange\.server\.port=/ {gsub(/[[:space:]]/, "", $2); print $2; exit}' "${config}")" + if [[ -n "${configured_exchange_port}" && "${configured_exchange_port}" != "${exchange_port}" ]]; then + echo_error "Worker ${worker_id}: configured exchange port ${configured_exchange_port} does not match HTTP+3 (${exchange_port})" + fi + specs+="${specs:+,}worker${worker_id}-http:${http_port},worker${worker_id}-exchange:${exchange_port}" + worker_id=$((worker_id + 1)) + done + CLUSTER_PORT_SPECS_BY_NODE["${node}"]="${specs}" + log_file="${LOGS}/port_preflight_${node}.log" + if ! run_port_inspection "${node}" "${specs}" require-free "${log_file}" "${preflight_timeout}"; then + cat "${log_file}" >&2 + echo_error "Fixed Presto/UCX ports are not bindable on ${node}; see ${log_file}" + fi + echo "Port preflight passed on ${node}: ${specs}" + done +} + +function collect_startup_diagnostics { + local reason="${1:-startup failure}" node specs log_file index + echo "Collecting cluster startup diagnostics: ${reason}" >&2 + echo "---- coord.log (last 120 lines) ----" >&2 + tail -n 120 "${LOGS}/coord.log" 2>/dev/null >&2 || true + for ((index = 0; index < ${#WORKER_SRUN_IDS[@]}; ++index)); do + echo "---- worker_${WORKER_SRUN_IDS[${index}]}.log on ${WORKER_SRUN_NODES[${index}]} (last 120 lines) ----" >&2 + tail -n 120 "${LOGS}/worker_${WORKER_SRUN_IDS[${index}]}.log" 2>/dev/null >&2 || true + done + for node in $(scontrol show hostnames "${SLURM_JOB_NODELIST}"); do + specs="${CLUSTER_PORT_SPECS_BY_NODE["${node}"]:-}" + [[ -n "${specs}" ]] || continue + log_file="${LOGS}/startup_ports_${node}.log" + run_port_inspection "${node}" "${specs}" report "${log_file}" 0 || true + echo "---- listener ownership on ${node} ----" >&2 + cat "${log_file}" >&2 || true + done +} + +# Stop exactly the coordinator and worker srun clients started above, wait for +# their steps to disappear, and verify that their fixed ports have been +# released before the batch allocation exits. +function stop_cluster { + [[ "${CLUSTER_STOPPED}" == "0" ]] || return 0 + CLUSTER_STOPPED=1 + + local stop_timeout="${PRESTO_CLUSTER_STOP_TIMEOUT:-30}" + local port_timeout="${PRESTO_PORT_RELEASE_TIMEOUT:-90}" + [[ "${stop_timeout}" =~ ^[0-9]+$ ]] || { + echo "Invalid PRESTO_CLUSTER_STOP_TIMEOUT=${stop_timeout}" >&2 + return 1 + } + [[ "${port_timeout}" =~ ^[0-9]+$ ]] || { + echo "Invalid PRESTO_PORT_RELEASE_TIMEOUT=${port_timeout}" >&2 + return 1 + } + + local -a tracked_pids=() + local pid index deadline any_running=0 cleanup_rc=0 node specs log_file + for ((index = ${#WORKER_SRUN_PIDS[@]} - 1; index >= 0; --index)); do + pid="${WORKER_SRUN_PIDS[${index}]}" + [[ -n "${pid}" ]] || continue + tracked_pids+=("${pid}") + if tracked_process_is_running "${pid}"; then + echo "Stopping worker ${WORKER_SRUN_IDS[${index}]} srun PID ${pid}..." + kill -TERM "${pid}" 2>/dev/null || true + fi + done + if [[ -n "${COORD_SRUN_PID}" ]]; then + tracked_pids+=("${COORD_SRUN_PID}") + if tracked_process_is_running "${COORD_SRUN_PID}"; then + echo "Stopping coordinator srun PID ${COORD_SRUN_PID}..." + kill -TERM "${COORD_SRUN_PID}" 2>/dev/null || true + fi + fi + + deadline=$((SECONDS + stop_timeout)) + while true; do + any_running=0 + for pid in "${tracked_pids[@]}"; do + if tracked_process_is_running "${pid}"; then + any_running=1 + break + fi + done + (( any_running == 0 || SECONDS >= deadline )) && break + sleep 1 + done + + if (( any_running != 0 )); then + echo "Cluster steps did not stop within ${stop_timeout}s; sending KILL to the tracked srun clients." >&2 + for pid in "${tracked_pids[@]}"; do + tracked_process_is_running "${pid}" && kill -KILL "${pid}" 2>/dev/null || true + done + cleanup_rc=1 + fi + for pid in "${tracked_pids[@]}"; do + wait "${pid}" 2>/dev/null || true + done + COORD_SRUN_PID="" + WORKER_SRUN_PIDS=() + + for node in $(scontrol show hostnames "${SLURM_JOB_NODELIST}"); do + specs="${CLUSTER_PORT_SPECS_BY_NODE["${node}"]:-}" + [[ -n "${specs}" ]] || continue + log_file="${LOGS}/port_cleanup_${node}.log" + if ! run_port_inspection "${node}" "${specs}" require-free "${log_file}" "${port_timeout}"; then + echo "Ports remained non-bindable after tracked-step cleanup on ${node}:" >&2 + cat "${log_file}" >&2 || true + cleanup_rc=1 + else + echo "Cluster ports released on ${node}." + fi + done + + return "${cleanup_rc}" +} + +# When requested by launch-run.sh --verify-cpu-ucx, require every expected +# HTTP+3 UCX port to be claimed before query submission. Do not infer readiness +# from UCX log strings: their presence depends on log level and UCX version, +# which made the old verifier reject healthy, fully registered clusters. +function verify_cpu_ucx_worker_startup { + [[ "${VERIFY_CPU_UCX:-0}" == "1" && "${VARIANT_TYPE:-gpu}" == "cpu" ]] || return 0 + + local ready_timeout="${PRESTO_CPU_UCX_READY_TIMEOUT:-60}" + [[ "${ready_timeout}" =~ ^[0-9]+$ ]] || \ + echo_error "Invalid PRESTO_CPU_UCX_READY_TIMEOUT=${ready_timeout}" + + local node specs exchange_specs entry role log_file + local -a entries + for node in $(scontrol show hostnames "${SLURM_JOB_NODELIST}"); do + specs="${CLUSTER_PORT_SPECS_BY_NODE["${node}"]:-}" + exchange_specs="" + IFS=',' read -ra entries <<< "${specs}" + for entry in "${entries[@]}"; do + role="${entry%%:*}" + [[ "${role}" == worker*-exchange ]] || continue + exchange_specs+="${exchange_specs:+,}${entry}" + done + [[ -n "${exchange_specs}" ]] || \ + echo_error "CPU UCX verification found no expected exchange ports on ${node}" + + log_file="${LOGS}/cpu_ucx_readiness_${node}.log" + if ! run_port_inspection "${node}" "${exchange_specs}" require-bound "${log_file}" "${ready_timeout}"; then + cat "${log_file}" >&2 || true + collect_startup_diagnostics "CPU UCX listener readiness failed on ${node}" + echo_error "CPU UCX listener readiness verification failed on ${node}; see ${log_file}" + fi + cat "${log_file}" + done + echo "CPU UCX listener ports verified on all ${NUM_WORKERS} worker(s)." +} + +# For multi-worker runs, require a retained coordinator query to report at +# least one CPU UCX replacement operator. Startup verification above proves +# every listener was ready before query submission; this check proves the +# selected query actually used the CPU-row UCX path. A one-worker plan cannot +# prove inter-worker transport and is therefore limited to listener readiness. +function verify_cpu_ucx_runtime { + [[ "${VERIFY_CPU_UCX:-0}" == "1" && "${VARIANT_TYPE:-gpu}" == "cpu" ]] || return 0 + + [[ "${NUM_WORKERS:-}" =~ ^[1-9][0-9]*$ ]] || \ + echo_error "CPU UCX verification requires a positive NUM_WORKERS value" + if (( NUM_WORKERS == 1 )); then + echo "CPU UCX runtime operator verification skipped for one worker; listener readiness was verified before query submission." + return 0 + fi + + local query_details_dir="${RESULT_DIR:-${SCRIPT_DIR}/result_dir}/coordinator_queries/queries" + if ! grep -R -E -q \ + '"operatorType"[[:space:]]*:[[:space:]]*"UcxCpuRow(Exchange|PartitionedOutput)"' \ + "${query_details_dir}" 2>/dev/null; then + echo "CPU UCX verification found no UcxCpuRowExchange or UcxCpuRowPartitionedOutput operator." >&2 + echo "Run an exchange-heavy query (for example TPC-H Q18) with at least two workers." >&2 + echo_error "CPU UCX operator verification failed" + fi + echo "CPU UCX replacement operator verified in retained query details." +} + # Start the coordinator, wait for it to come up, fan out NUM_GPUS_PER_NODE -# workers per node across $SLURM_JOB_NODELIST, then block until all +# local worker slots per node across $SLURM_JOB_NODELIST, then block until all # $NUM_WORKERS register. Used by both the benchmark and analyze flows. function start_cluster { + prepare_cpu_numa_layout + preflight_cluster_ports + echo "Starting Presto coordinator on ${COORD}..." run_coordinator wait_until_coordinator_is_running echo "Starting ${NUM_WORKERS} Presto workers across ${NUM_NODES} nodes..." - local worker_id=0 node gpu_id + local worker_id=0 node local_worker_id for node in $(scontrol show hostnames "$SLURM_JOB_NODELIST"); do - for gpu_id in $(seq 0 $((NUM_GPUS_PER_NODE - 1))); do - echo " Starting worker ${worker_id} on node ${node} GPU ${gpu_id}" - run_worker "${gpu_id}" "$WORKER_IMAGE" "${node}" "$worker_id" + for local_worker_id in $(seq 0 $((NUM_GPUS_PER_NODE - 1))); do + if [[ "${VARIANT_TYPE}" == "gpu" ]]; then + echo " Starting worker ${worker_id} on node ${node} GPU ${local_worker_id}" + else + echo " Starting worker ${worker_id} on node ${node} local slot ${local_worker_id}" + fi + run_worker "${local_worker_id}" "$WORKER_IMAGE" "${node}" "$worker_id" worker_id=$((worker_id + 1)) done done echo "Waiting for ${NUM_WORKERS} workers to register with coordinator..." wait_for_workers_to_register "$NUM_WORKERS" + verify_cpu_ucx_worker_startup + start_perf_sampler } # ---------------------------------------------------------------------------- @@ -532,65 +1377,79 @@ function run_queries { # (VT_ROOT is bind-mounted as /workspace) so the path stays correct even # if this directory is renamed away from `presto-nvl72`. local container_script_dir="${SCRIPT_DIR/${VT_ROOT}//workspace}" + local result_dir="${RESULT_DIR:-${SCRIPT_DIR}/result_dir}" + local container_result_dir="${result_dir/${VT_ROOT}//workspace}" + [[ "${container_result_dir}" == /workspace/* ]] || \ + echo_error "RESULT_DIR must be below VT_ROOT (${VT_ROOT}) so it is mounted in the CLI container: ${result_dir}" + + source "${SCRIPT_DIR}/defaults.env" local extra_args=() [[ "${ENABLE_METRICS}" == "1" ]] && extra_args+=("-m") - [[ "${ENABLE_NSYS}" == "1" ]] && extra_args+=("-p" "--profile-script-path" "${container_script_dir}/profiler_functions.sh") + if [[ "${ENABLE_NSYS}" == "1" ]]; then + extra_args+=("-p" "--profile-script-path" "${container_script_dir}/profiler_functions.sh") + elif [[ "${ENABLE_PERF:-0}" == "1" ]]; then + extra_args+=("-p" "--profile-script-path" "${container_script_dir}/perf_profiler_functions.sh") + fi [[ -n "${QUERIES:-}" ]] && extra_args+=("-q" "${QUERIES}") - - source "${SCRIPT_DIR}/defaults.env" - - # The upstream coordinator image ships without jq, which - # run_benchmark.sh's wait_for_worker_node_registration requires. - # yum/dnf cannot install it at runtime because the container root is - # a read-only squashfs (/var/cache/dnf is read-only). Stage a - # statically-linked jq under VT_ROOT (which is bind-mounted into the - # container as /workspace) and prepend that to PATH. The download - # is cached across runs so the cost is paid once. - local jq_cache="${VT_ROOT}/.cache/bin" - local jq_arch - case "$(uname -m)" in - aarch64|arm64) jq_arch="arm64" ;; - x86_64|amd64) jq_arch="amd64" ;; - *) echo_error "unsupported arch for jq download: $(uname -m)" ;; - esac - if [ ! -x "${jq_cache}/jq" ]; then - echo "Staging static jq (${jq_arch}) at ${jq_cache}/jq" - mkdir -p "${jq_cache}" - curl -sSL "https://github.com/jqlang/jq/releases/download/jq-1.7.1/jq-linux-${jq_arch}" \ - -o "${jq_cache}/jq" - chmod +x "${jq_cache}/jq" + if [[ -n "${QUERIES_FILE:-}" ]]; then + local container_queries_file="${QUERIES_FILE}" + if [[ "${container_queries_file}" == "${VT_ROOT}/"* ]]; then + container_queries_file="/workspace/${container_queries_file#"${VT_ROOT}/"}" + fi + [[ "${container_queries_file}" == /workspace/* ]] || \ + echo_error "QUERIES_FILE must be below VT_ROOT (${VT_ROOT}) so it is mounted in the CLI container: ${QUERIES_FILE}" + extra_args+=("--queries-file" "${container_queries_file}") fi + local benchmark_args=( + ./run_benchmark.sh + -b tpch + -s "tpchsf${scale_factor}" + -i "${num_iterations}" + "${extra_args[@]}" + --hostname "${COORD}" + --port "${PORT}" + -o "${container_result_dir}" + --skip-drop-cache + ) + local benchmark_cmd + printf -v benchmark_cmd '%q ' "${benchmark_args[@]}" + # Result validation is intentionally not wired here yet: that belongs # to PR #275 (upstream validate_results.py) and will be hooked up # after the PR merges and this branch is rebased. # Cache-drop is skipped because it requires docker (not available on # the cluster). - run_coord_image "export PATH=/workspace/.cache/bin:\$PATH; \ - export PORT=$PORT; \ + run_coord_image "export PORT=$PORT; \ export HOSTNAME=$COORD; \ export PRESTO_DATA_DIR=/var/lib/presto/data/hive/data/user_data; \ export MINIFORGE_HOME=/workspace/miniforge3; \ export HOME=/workspace; \ cd /workspace/presto/scripts; \ - ./run_benchmark.sh -b tpch -s tpchsf${scale_factor} -i ${num_iterations} ${extra_args[*]} \ - --hostname ${COORD} --port $PORT -o ${container_script_dir}/result_dir --skip-drop-cache" "cli" + ${benchmark_cmd}" "cli" } # Check if the coordinator is running via curl. Fail after 10 retries. function wait_until_coordinator_is_running { echo "waiting for coordinator to be accessible" validate_environment_preconditions COORD LOGS - local state="INACTIVE" + local state="INACTIVE" coord_rc=0 for i in {1..10}; do state=$(curl -s http://${COORD}:${PORT}/v1/info/state || true) if [[ "$state" == "\"ACTIVE\"" ]]; then echo "coord started. state: $state" return 0 fi + if [[ -n "${COORD_SRUN_PID}" ]] && ! tracked_process_is_running "${COORD_SRUN_PID}"; then + wait "${COORD_SRUN_PID}" || coord_rc=$? + COORD_SRUN_PID="" + collect_startup_diagnostics "coordinator srun exited with status ${coord_rc}" + echo_error "coord did not start; its srun exited with status ${coord_rc}" + fi sleep 5 done + collect_startup_diagnostics "coordinator did not become ACTIVE" echo_error "coord did not start. state: $state" } @@ -600,15 +1459,29 @@ function wait_for_workers_to_register { [ $# -ne 1 ] && echo_error "$0 expected one argument for 'expected number of workers'" echo "waiting for $1 workers to register" local expected_num_workers=$1 - local num_workers=0 + local num_workers=0 index pid worker_rc for i in {1..60}; do - num_workers=$(curl -s http://${COORD}:${PORT}/v1/node | jq length) - if (( $num_workers == $expected_num_workers )); then + num_workers=$(curl -fsS "http://${COORD}:${PORT}/v1/node" 2>/dev/null \ + | python3 -c 'import json, sys; print(len(json.load(sys.stdin)))' 2>/dev/null \ + || echo 0) + if [[ "${num_workers}" =~ ^[0-9]+$ ]] && (( num_workers == expected_num_workers )); then echo "workers registered. num_nodes: $num_workers" return 0 fi + for ((index = 0; index < ${#WORKER_SRUN_PIDS[@]}; ++index)); do + pid="${WORKER_SRUN_PIDS[${index}]}" + [[ -n "${pid}" ]] || continue + if ! tracked_process_is_running "${pid}"; then + worker_rc=0 + wait "${pid}" || worker_rc=$? + WORKER_SRUN_PIDS[${index}]="" + collect_startup_diagnostics "worker ${WORKER_SRUN_IDS[${index}]} on ${WORKER_SRUN_NODES[${index}]} exited with status ${worker_rc}; ${num_workers}/${expected_num_workers} workers registered" + echo_error "worker ${WORKER_SRUN_IDS[${index}]} exited before cluster registration completed" + fi + done sleep 5 done + collect_startup_diagnostics "worker registration timed out at ${num_workers}/${expected_num_workers}" echo_error "workers failed to register. num_nodes: $num_workers" } @@ -630,7 +1503,7 @@ function validate_config_directory { } function collect_results { - local result_dir="${SCRIPT_DIR}/result_dir" + local result_dir="${RESULT_DIR:-${SCRIPT_DIR}/result_dir}" echo "Copying configs to ${result_dir}/configs/..." mkdir -p "${result_dir}/configs" @@ -643,7 +1516,7 @@ function collect_results { } function inject_benchmark_metadata { - local result_file="${SCRIPT_DIR}/result_dir/benchmark_result.json" + local result_file="${RESULT_DIR:-${SCRIPT_DIR}/result_dir}/benchmark_result.json" if [ ! -f "${result_file}" ]; then echo "Warning: ${result_file} not found, skipping metadata injection" return @@ -685,32 +1558,60 @@ function inject_benchmark_metadata { image_digest="${image_digest:-unknown}" echo "Image digest: ${image_digest}" - local tmp_file - tmp_file=$(mktemp) - jq --arg kind "$kind" \ - --arg timestamp "$timestamp" \ - --argjson n_workers "$NUM_WORKERS" \ - --argjson node_count "$NUM_NODES" \ - --argjson scale_factor "$SCALE_FACTOR" \ - --argjson gpu_count "$gpu_count" \ - --arg gpu_name "$gpu_name" \ - --argjson num_drivers "$num_drivers" \ - --arg worker_image "$WORKER_IMAGE" \ - --arg image_digest "$image_digest" \ - --arg engine "$engine" \ - '.context += { - kind: $kind, - timestamp: $timestamp, - n_workers: $n_workers, - node_count: $node_count, - scale_factor: $scale_factor, - gpu_count: $gpu_count, - gpu_name: $gpu_name, - num_drivers: $num_drivers, - worker_image: $worker_image, - image_digest: $image_digest, - engine: $engine - }' "${result_file}" > "${tmp_file}" && mv "${tmp_file}" "${result_file}" + python3 - "${result_file}" \ + "${kind}" "${timestamp}" "${NUM_WORKERS}" "${NUM_NODES}" \ + "${SCALE_FACTOR}" "${gpu_count}" "${gpu_name}" "${num_drivers}" \ + "${WORKER_IMAGE}" "${image_digest}" "${engine}" <<'PY' +import json +import os +import pathlib +import sys + +( + result_path, + kind, + timestamp, + n_workers, + node_count, + scale_factor, + gpu_count, + gpu_name, + num_drivers, + worker_image, + image_digest, + engine, +) = sys.argv[1:] + +path = pathlib.Path(result_path) +with path.open() as source: + result = json.load(source) + +context = result.get("context") +if not isinstance(context, dict): + context = {} + result["context"] = context +context.update( + { + "kind": kind, + "timestamp": timestamp, + "n_workers": int(n_workers), + "node_count": int(node_count), + "scale_factor": int(scale_factor), + "gpu_count": int(gpu_count), + "gpu_name": gpu_name, + "num_drivers": int(num_drivers), + "worker_image": worker_image, + "image_digest": image_digest, + "engine": engine, + } +) + +temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") +with temporary.open("w") as output: + json.dump(result, output, indent=2) + output.write("\n") +os.replace(temporary, path) +PY echo "Injected benchmark metadata into ${result_file}" } @@ -772,7 +1673,7 @@ function wait_for_nsys_report_generation { sleep 5 done - echo "Copying nsys reports to ${SCRIPT_DIR}/result_dir/..." - cp "${LOGS}"/*.nsys-rep "${SCRIPT_DIR}/result_dir/" + echo "Copying nsys reports to ${RESULT_DIR:-${SCRIPT_DIR}/result_dir}/..." + cp "${LOGS}"/*.nsys-rep "${RESULT_DIR:-${SCRIPT_DIR}/result_dir}/" fi } diff --git a/presto/slurm/presto-nvl72/launch-analyze-tables.sh b/presto/slurm/presto-nvl72/launch-analyze-tables.sh index f006fb68..734713a2 100755 --- a/presto/slurm/presto-nvl72/launch-analyze-tables.sh +++ b/presto/slurm/presto-nvl72/launch-analyze-tables.sh @@ -16,6 +16,7 @@ # [-d|--data-dir ] # [-g|--num-workers-per-node ] # [-w|--worker-image ] [-c|--coord-image ] +# [--worker-env-file ] # [additional sbatch options] # # Examples: @@ -68,6 +69,7 @@ while [[ $# -gt 0 ]]; do -g|--num-workers-per-node) requires_value "$1" "${2:-}"; NUM_GPUS_PER_NODE="$2"; shift 2 ;; -w|--worker-image) requires_value "$1" "${2:-}"; WORKER_IMAGE="$2"; shift 2 ;; -c|--coord-image) requires_value "$1" "${2:-}"; COORD_IMAGE="$2"; shift 2 ;; + --worker-env-file) requires_value "$1" "${2:-}"; WORKER_ENV_FILE="$2"; shift 2 ;; -h|--help) usage; exit 0 ;; --) shift; EXTRA_ARGS+=("$@"); break ;; *) EXTRA_ARGS+=("$1"); shift ;; @@ -99,6 +101,10 @@ resolve_cluster_variant cpu build_cluster_sbatch_args "${CLUSTER_TIME_ANALYZE}" # Pre-flight: verify prerequisites before queueing the job. +preflight_file "${WORKER_ENV_FILE}" "worker environment" \ + "Set WORKER_ENV_FILE or pass --worker-env-file " +WORKER_ENV_FILE="$(canonicalize_file_path "${WORKER_ENV_FILE}")" +preflight_image_roles "${WORKER_IMAGE}" "${COORD_IMAGE}" preflight_image "${WORKER_IMAGE}" \ "Pull it (see ./pull_ghcr_image.sh) or override with -w " preflight_image "${COORD_IMAGE}" \ diff --git a/presto/slurm/presto-nvl72/launch-run.sh b/presto/slurm/presto-nvl72/launch-run.sh index 92807312..b6d63516 100755 --- a/presto/slurm/presto-nvl72/launch-run.sh +++ b/presto/slurm/presto-nvl72/launch-run.sh @@ -14,7 +14,10 @@ # [-i|--iterations ] [--cpu] [-g|--num-workers-per-node ] # [-w|--worker-image ] [-c|--coord-image ] # [-o|--output-path ] [-q|--queries ] -# [--disable-gds] [-m|--metrics] [-p|--profile] +# [--queries-file ] +# [--verify-cpu-ucx] +# [--legacy-cpu-numa] +# [--disable-gds] [-m|--metrics] [-p|--profile] [--perf] # [additional sbatch options] # ============================================================================== @@ -42,8 +45,26 @@ SCRIPT_DIR="$PWD" ENABLE_GDS=1 ENABLE_METRICS=0 ENABLE_NSYS=0 -NSYS_WORKER_ID=0 +ENABLE_PERF=0 +NSYS_WORKER_IDS="0" +PROFILE_ITERATIONS="" # comma-separated iter indices; empty = combined per query +NSYS_LAUNCH_OPTS="-t nvtx,cuda" # trace flags passed to `nsys launch` +PERF_WORKER_ID=0 +PERF_FREQUENCY="${PERF_FREQUENCY:-19}" +PERF_EVENT="${PERF_EVENT:-cpu-clock:u}" +PERF_CALL_GRAPH="${PERF_CALL_GRAPH:-dwarf,8192}" +PERF_USER_REGS="${PERF_USER_REGS:-auto}" +PERF_NOFILE_LIMIT="${PERF_NOFILE_LIMIT:-65536}" +PERF_THREADS_PER_SHARD="${PERF_THREADS_PER_SHARD:-128}" +PERF_RECORD_STOP_TIMEOUT="${PERF_RECORD_STOP_TIMEOUT:-120}" +PERF_POSTPROCESS="${PERF_POSTPROCESS:-0}" +PERF_FINALIZE_TIMEOUT="${PERF_FINALIZE_TIMEOUT:-0}" QUERIES="" +QUERIES_FILE="" +CONFIG_OVERRIDES="" +# Deliberately CLI-only. A stale exported VERIFY_CPU_UCX must not silently turn +# verification on in a later run; --verify-cpu-ucx below is the only enabler. +VERIFY_CPU_UCX=0 usage() { cat < Override coordinator image from cluster config -o, --output-path Copy results into this directory after the run -q, --queries Comma-separated query filter (e.g. "1,6,21") + --queries-file Custom JSON query file passed to run_benchmark.sh --worker-env-file Override worker.env (default: ./worker.env) + --verify-cpu-ucx Assert every CPU UCX listener is ready before + queries; for multi-worker runs, also assert that + a query uses CPU UCX replacement operators. + Enables UCX diagnostics. + --config-overrides Semicolon-separated property overrides applied + after generate_configs (e.g. + "task.max-drivers-per-task=8;cudf.batch_size_min_threshold=200000") --cpu Use CPU partition/images (overrides cluster default) --gpu Use GPU partition/images (overrides cluster default) + --numa Enable NUMA pinning for workers --no-numa Disable NUMA pinning for workers --disable-gds Disable GPU Direct Storage -m, --metrics Enable metrics collection -p, --profile Enable nsys profiling - --nsys-worker-id Worker ID to profile (default: ${NSYS_WORKER_ID}) + --perf Sample worker 0 with host perf during each + selected query/profile iteration. Captures raw + perf.data plus a portable symfs under + result_dir_/profiles/perf/worker_0/. + --perf-frequency Sampling frequency (default: ${PERF_FREQUENCY}) + --perf-event perf event (default: ${PERF_EVENT}) + --perf-call-graph perf call-graph mode (default: ${PERF_CALL_GRAPH}) + --perf-user-regs Registers passed to perf --user-regs, or + 'auto'/'perf-default' (default: ${PERF_USER_REGS}) + --perf-nofile-limit Minimum host perf soft nofile limit + (default: ${PERF_NOFILE_LIMIT}) + --perf-threads-per-shard + Maximum target TIDs assigned to one perf + recorder (default: ${PERF_THREADS_PER_SHARD}) + --perf-record-stop-timeout + Maximum time for perf record to flush and stop + after a query (default: + ${PERF_RECORD_STOP_TIMEOUT}) + --perf-postprocess Generate reports and flamegraphs after all raw + captures are published. By default this expensive + work is deferred to process-perf-captures.sh. + --perf-finalize-timeout + Maximum wait for post-suite artifact publication; + 0 uses the Slurm job time limit (default: + ${PERF_FINALIZE_TIMEOUT}) + --nsys-worker-ids Worker IDs to profile: comma list (e.g. "0,3,5") or + 'all' to profile every worker (default: ${NSYS_WORKER_IDS}) + --nsys-worker-id Alias for --nsys-worker-ids accepting a single ID + --profile-iterations + Comma-separated 0-based iteration indices to + profile separately (e.g. "1" to skip iteration + 0, or "0,1" to capture both). Applies to nsys + and perf. When unset, one capture spans every + iteration of each selected query. + --nsys-launch-opts Options passed to \`nsys launch\` controlling + what is traced. Default: "-t nvtx,cuda". + Examples: + "-t nvtx,cuda,osrt" (add OS runtime) + "-t nvtx,ucx,osrt --sample=process-tree --backtrace=dwarf" + "-t nvtx,cuda --gpu-metrics-device=all" + "-t nvtx,cuda --cuda-memory-usage=true" -h, --help Show this help message and exit Any arguments after -- are passed directly to sbatch. @@ -90,14 +160,47 @@ while [[ $# -gt 0 ]]; do -c|--coord-image) requires_value "$1" "${2:-}"; COORD_IMAGE="$2"; shift 2 ;; -o|--output-path) requires_value "$1" "${2:-}"; OUTPUT_PATH="$2"; shift 2 ;; -q|--queries) requires_value "$1" "${2:-}"; QUERIES="$2"; shift 2 ;; + --queries-file) requires_value "$1" "${2:-}"; QUERIES_FILE="$2"; shift 2 ;; --worker-env-file) requires_value "$1" "${2:-}"; WORKER_ENV_FILE="$2"; shift 2 ;; - --nsys-worker-id) requires_value "$1" "${2:-}"; NSYS_WORKER_ID="$2"; shift 2 ;; + --verify-cpu-ucx) VERIFY_CPU_UCX=1; shift ;; + --config-overrides) requires_value "$1" "${2:-}"; CONFIG_OVERRIDES="$2"; shift 2 ;; + --nsys-worker-id) requires_value "$1" "${2:-}"; NSYS_WORKER_IDS="$2"; shift 2 ;; + --nsys-worker-ids) requires_value "$1" "${2:-}"; NSYS_WORKER_IDS="$2"; shift 2 ;; + --profile-iterations) requires_value "$1" "${2:-}"; PROFILE_ITERATIONS="$2"; shift 2 ;; + --nsys-launch-opts) + # Can't use requires_value here — it disallows dash-prefixed values, + # but valid nsys flags like "-t nvtx,ucx --sample=process-tree" start with -. + [[ -n "${2:-}" ]] || { echo "Error: $1 requires a value" >&2; exit 1; } + NSYS_LAUNCH_OPTS="$2"; shift 2 ;; --cpu) VARIANT_TYPE="cpu"; shift ;; --gpu) VARIANT_TYPE="gpu"; shift ;; + --numa) USE_NUMA="1"; shift ;; --no-numa) USE_NUMA="0"; shift ;; --disable-gds) ENABLE_GDS=0; shift ;; -m|--metrics) ENABLE_METRICS=1; shift ;; -p|--profile) ENABLE_NSYS=1; shift ;; + --perf) ENABLE_PERF=1; shift ;; + --perf-frequency) requires_value "$1" "${2:-}"; PERF_FREQUENCY="$2"; shift 2 ;; + --perf-event) requires_value "$1" "${2:-}"; PERF_EVENT="$2"; shift 2 ;; + --perf-call-graph) + [[ -n "${2:-}" ]] || { echo "Error: $1 requires a value" >&2; exit 1; } + PERF_CALL_GRAPH="$2"; shift 2 ;; + --perf-user-regs) + requires_value "$1" "${2:-}" + PERF_USER_REGS="$2"; shift 2 ;; + --perf-nofile-limit) + requires_value "$1" "${2:-}" + PERF_NOFILE_LIMIT="$2"; shift 2 ;; + --perf-threads-per-shard) + requires_value "$1" "${2:-}" + PERF_THREADS_PER_SHARD="$2"; shift 2 ;; + --perf-record-stop-timeout) + requires_value "$1" "${2:-}" + PERF_RECORD_STOP_TIMEOUT="$2"; shift 2 ;; + --perf-postprocess) PERF_POSTPROCESS=1; shift ;; + --perf-finalize-timeout) + requires_value "$1" "${2:-}" + PERF_FINALIZE_TIMEOUT="$2"; shift 2 ;; -h|--help) usage; exit 0 ;; --) shift; EXTRA_ARGS+=("$@"); break ;; *) EXTRA_ARGS+=("$1"); shift ;; @@ -106,12 +209,52 @@ done [[ -z "${NODES_COUNT}" ]] && { echo "Error: -n|--nodes is required (see --help)" >&2; exit 1; } [[ -z "${SCALE_FACTOR}" ]] && { echo "Error: -s|--scale-factor is required (see --help)" >&2; exit 1; } - -# Clean up old output files — use rm -rf so subdirectories (e.g. query_results/) -# are fully removed and stale benchmark_result.json cannot survive a cancelled run. -rm -rf result_dir logs 2>/dev/null || true -rm -f *.out *.err 2>/dev/null || true -mkdir -p result_dir logs +if [[ -n "${PROFILE_ITERATIONS}" ]] && ! [[ "${PROFILE_ITERATIONS}" =~ ^[0-9]+(,[0-9]+)*$ ]]; then + echo "Error: --profile-iterations expects a comma-separated list of 0-based ints; got '${PROFILE_ITERATIONS}'" >&2 + exit 1 +fi +if [[ "${ENABLE_NSYS}" == "1" && "${ENABLE_PERF}" == "1" ]]; then + echo "Error: --profile (nsys) and --perf must run in separate jobs" >&2 + exit 1 +fi +if [[ "${PERF_POSTPROCESS}" == "1" && "${ENABLE_PERF}" != "1" ]]; then + echo "Error: --perf-postprocess requires --perf" >&2 + exit 1 +fi +if ! [[ "${PERF_FREQUENCY}" =~ ^[1-9][0-9]*$ ]]; then + echo "Error: --perf-frequency must be a positive integer; got '${PERF_FREQUENCY}'" >&2 + exit 1 +fi +if ! [[ "${PERF_NOFILE_LIMIT}" =~ ^[1-9][0-9]*$ ]]; then + echo "Error: --perf-nofile-limit must be a positive integer; got '${PERF_NOFILE_LIMIT}'" >&2 + exit 1 +fi +if ! [[ "${PERF_THREADS_PER_SHARD}" =~ ^[1-9][0-9]*$ ]]; then + echo "Error: --perf-threads-per-shard must be a positive integer; got '${PERF_THREADS_PER_SHARD}'" >&2 + exit 1 +fi +if ! [[ "${PERF_RECORD_STOP_TIMEOUT}" =~ ^[1-9][0-9]*$ ]]; then + echo "Error: --perf-record-stop-timeout must be a positive integer; got '${PERF_RECORD_STOP_TIMEOUT}'" >&2 + exit 1 +fi +if ! [[ "${PERF_POSTPROCESS}" =~ ^[01]$ ]]; then + echo "Error: PERF_POSTPROCESS must be 0 or 1; got '${PERF_POSTPROCESS}'" >&2 + exit 1 +fi +if ! [[ "${PERF_FINALIZE_TIMEOUT}" =~ ^[0-9]+$ ]]; then + echo "Error: --perf-finalize-timeout must be a non-negative integer; got '${PERF_FINALIZE_TIMEOUT}'" >&2 + exit 1 +fi +if [[ -n "${PROFILE_ITERATIONS}" ]]; then + IFS=',' read -ra _profile_iteration_list <<< "${PROFILE_ITERATIONS}" + for _profile_iteration in "${_profile_iteration_list[@]}"; do + if (( _profile_iteration >= NUM_ITERATIONS )); then + echo "Error: profile iteration ${_profile_iteration} is outside -i ${NUM_ITERATIONS} (indices are 0-based)" >&2 + exit 1 + fi + done + unset _profile_iteration _profile_iteration_list +fi echo "Submitting Presto TPC-H benchmark job..." echo "" @@ -124,6 +267,27 @@ resolve_cluster_variant "${VARIANT_TYPE}" : "${NUM_GPUS_PER_NODE:=${CLUSTER_NUM_WORKERS_PER_NODE:-}}" : "${USE_NUMA:=${CLUSTER_USE_NUMA:-0}}" +# GDS is a GPU-only data path. CPU launches should not require --disable-gds +# and should never expose cufile settings to a CPU worker image. +if [[ "${VARIANT_TYPE}" == "cpu" ]]; then + ENABLE_GDS=0 +fi + +if [[ "${VERIFY_CPU_UCX}" == "1" ]]; then + [[ "${VARIANT_TYPE}" == "cpu" ]] || { echo "Error: --verify-cpu-ucx requires --cpu" >&2; exit 1; } + export UCX_LOG_LEVEL="${UCX_LOG_LEVEL:-info}" + export UCX_PROTO_INFO="${UCX_PROTO_INFO:-y}" +fi + +# Expand --nsys-worker-ids=all to the full 0..N-1 range now that NUM_GPUS_PER_NODE +# is known. Done here (not at arg-parse time) because total worker count depends +# on cluster-resolved values. +if [[ "${NSYS_WORKER_IDS}" == "all" ]]; then + _total_workers=$(( NODES_COUNT * NUM_GPUS_PER_NODE )) + NSYS_WORKER_IDS="$(seq -s, 0 $(( _total_workers - 1 )))" + unset _total_workers +fi + # Validate required values before submitting VTYPE_UPPER="${VARIANT_TYPE^^}" [[ -z "${WORKER_IMAGE}" ]] && { echo "Error: worker image not set — set CLUSTER_${VTYPE_UPPER}_DEFAULT_WORKER_IMAGE in cluster_config.env or pass -w"; exit 1; } @@ -138,6 +302,25 @@ build_cluster_sbatch_args "${CLUSTER_TIME_BENCHMARK}" # Pre-flight: verify prerequisites before queueing the job. ANALYZE_HINT="./launch-analyze-tables.sh -s ${SCALE_FACTOR}" +preflight_file "${WORKER_ENV_FILE}" "worker environment" \ + "Set WORKER_ENV_FILE or pass --worker-env-file " +WORKER_ENV_FILE="$(canonicalize_file_path "${WORKER_ENV_FILE}")" + +if [[ -n "${QUERIES_FILE}" ]]; then + preflight_file "${QUERIES_FILE}" "queries JSON" + QUERIES_FILE="$(canonicalize_file_path "${QUERIES_FILE}")" + if [[ "${QUERIES_FILE}" != "${VT_ROOT}/"* ]]; then + echo "Error: --queries-file must be inside ${VT_ROOT} so it is available in the CLI container." >&2 + echo " Got: ${QUERIES_FILE}" >&2 + exit 1 + fi + if ! python3 -m json.tool "${QUERIES_FILE}" >/dev/null 2>&1; then + echo "Error: queries file is not valid JSON: ${QUERIES_FILE}" >&2 + exit 1 + fi + QUERIES_FILE="/workspace/${QUERIES_FILE#"${VT_ROOT}/"}" +fi +preflight_image_roles "${WORKER_IMAGE}" "${COORD_IMAGE}" preflight_image "${WORKER_IMAGE}" \ "Pull it (see ./pull_ghcr_image.sh) or override with -w " preflight_image "${COORD_IMAGE}" \ @@ -146,6 +329,13 @@ preflight_dir "${DATA}/tpch-rs-${SCALE_FACTOR}" "TPC-H SF${SCALE_FACTOR} data" \ "./launch-gen-data.sh -s ${SCALE_FACTOR} -o ${DATA}/tpch-rs-${SCALE_FACTOR}" preflight_metastore "${SCALE_FACTOR}" "${ANALYZE_HINT}" +# Only transient live logs are reused. Completed result directories are named +# with their Slurm job ID and are never removed by a later launch, so an scp or +# rsync of an earlier run cannot race this cleanup. +rm -rf logs 2>/dev/null || true +rm -f *.out *.err 2>/dev/null || true +mkdir -p logs + # Submit job (include nodes/SF/iterations in file names) OUT_FMT="logs/presto-tpch-run_n${NODES_COUNT}_sf${SCALE_FACTOR}_i${NUM_ITERATIONS}_%j.out" ERR_FMT="logs/presto-tpch-run_n${NODES_COUNT}_sf${SCALE_FACTOR}_i${NUM_ITERATIONS}_%j.err" @@ -162,23 +352,61 @@ GRES_OPT=$([[ "$VARIANT_TYPE" == "gpu" ]] && echo "--gres=gpu:${NUM_GPUS_PER_NOD build_common_export_vars EXPORT_VARS+=",NUM_ITERATIONS=${NUM_ITERATIONS}" EXPORT_VARS+=",ENABLE_GDS=${ENABLE_GDS},ENABLE_METRICS=${ENABLE_METRICS}" -EXPORT_VARS+=",ENABLE_NSYS=${ENABLE_NSYS},NSYS_WORKER_ID=${NSYS_WORKER_ID}" -# Comma-separated query list can't ride EXPORT_VARS (comma is the separator); -# export it so sbatch picks it up via the ALL inheritance. +EXPORT_VARS+=",ENABLE_NSYS=${ENABLE_NSYS}" +EXPORT_VARS+=",ENABLE_PERF=${ENABLE_PERF},PERF_WORKER_ID=${PERF_WORKER_ID}" +EXPORT_VARS+=",VERIFY_CPU_UCX=${VERIFY_CPU_UCX}" +# Comma-separated values can't ride EXPORT_VARS (comma is the sbatch --export +# field separator); export them so sbatch picks them up via the ALL inheritance. [[ -n "${QUERIES}" ]] && export QUERIES +[[ -n "${QUERIES_FILE}" ]] && export QUERIES_FILE +export NSYS_WORKER_IDS +[[ -n "${PROFILE_ITERATIONS}" ]] && export PROFILE_ITERATIONS +# NSYS_LAUNCH_OPTS may contain spaces / equals signs / commas — ride ALL +# inheritance rather than EXPORT_VARS so the whole quoted string survives. +export NSYS_LAUNCH_OPTS +# PERF_CALL_GRAPH and PERF_USER_REGS contain commas, and the remaining values +# may contain perf punctuation. Inherit them through --export=ALL instead of +# splitting the sbatch export list. +export PERF_FREQUENCY PERF_EVENT PERF_CALL_GRAPH PERF_USER_REGS +export PERF_NOFILE_LIMIT PERF_THREADS_PER_SHARD PERF_RECORD_STOP_TIMEOUT +export PERF_POSTPROCESS PERF_FINALIZE_TIMEOUT +# CONFIG_OVERRIDES uses ';' internally but values may contain commas (e.g. +# "32MB,64MB" would break, but more pressingly any later additions); easiest +# to keep it off EXPORT_VARS entirely. +[[ -n "${CONFIG_OVERRIDES}" ]] && export CONFIG_OVERRIDES JOB_ID=$(sbatch --job-name="${JOB_NAME}" --nodes="${NODES_COUNT}" "${NODELIST_ARG[@]}" \ "${CLUSTER_SBATCH_ARGS[@]}" \ --export="${EXPORT_VARS}" \ --output="${OUT_FMT}" --error="${ERR_FMT}" "${EXTRA_ARGS[@]}" ${GRES_OPT} \ -run-presto-benchmarks.slurm | awk '{print $NF}') + run-presto-benchmarks.slurm | awk '{print $NF}') +RUN_RESULT_DIR="result_dir_${JOB_ID}" OUT_FILE="${OUT_FMT//%j/${JOB_ID}}" ERR_FILE="${ERR_FMT//%j/${JOB_ID}}" +# A text pointer is safe for automation: callers resolve it once and then copy +# the immutable job-specific directory. Unlike a mutable result_dir symlink, it +# cannot redirect individual files to a newer run midway through rsync/scp. +LATEST_RESULT_POINTER="${SCRIPT_DIR}/latest_result_dir.txt" +LATEST_RESULT_POINTER_TMP="${LATEST_RESULT_POINTER}.tmp.$$" +printf '%s\n' "${RUN_RESULT_DIR}" > "${LATEST_RESULT_POINTER_TMP}" +mv -f "${LATEST_RESULT_POINTER_TMP}" "${LATEST_RESULT_POINTER}" + +echo "Job submitted with ID: $JOB_ID" +echo "" + # Resolve and print first node IP once nodes are allocated echo "Resolving first node IP..." for i in {1..60}; do STATE=$(squeue -j "$JOB_ID" -h -o "%T" 2>/dev/null || true) NODELIST=$(squeue -j "$JOB_ID" -h -o "%N" 2>/dev/null || true) + if [[ -z "${STATE:-}" ]]; then + echo "Job ${JOB_ID} is no longer in squeue before node allocation." + break + fi + if [[ "${STATE}" =~ ^(CANCELLED|FAILED|TIMEOUT|COMPLETED|NODE_FAIL|PREEMPTED|BOOT_FAIL|DEADLINE|OUT_OF_MEMORY)$ ]]; then + echo "Job ${JOB_ID} reached state ${STATE} before node allocation." + break + fi if [[ -n "${NODELIST:-}" && "${NODELIST}" != "(null)" ]]; then FIRST_NODE=$(scontrol show hostnames "$NODELIST" | head -n 1) if [[ -n "${FIRST_NODE:-}" ]]; then @@ -198,8 +426,6 @@ for i in {1..60}; do sleep 5 done -echo "Job submitted with ID: $JOB_ID" -echo "" print_monitor_hints "${JOB_ID}" "${OUT_FILE}" "${ERR_FILE}" \ "tail -f logs/coord.log" \ "tail -f logs/worker_*.log" \ @@ -212,12 +438,21 @@ echo "" echo "Output files:" ls -lh "${OUT_FILE}" "${ERR_FILE}" 2>/dev/null || echo "No output files found" show_job_output "${OUT_FILE}" "${ERR_FILE}" "logs/cli.log" "benchmark results" -[[ "${JOB_STATE}" == "COMPLETED" ]] || exit 1 -if [[ -n "${OUTPUT_PATH}" ]]; then +if [[ -d "${RUN_RESULT_DIR}" ]]; then + echo "" + echo "Job results preserved at: ${SCRIPT_DIR}/${RUN_RESULT_DIR}" + echo "Latest-result pointer: ${LATEST_RESULT_POINTER}" +else + echo "No job result directory was created at ${SCRIPT_DIR}/${RUN_RESULT_DIR}" >&2 +fi + +if [[ -n "${OUTPUT_PATH}" && -d "${RUN_RESULT_DIR}" ]]; then echo "" echo "Copying results to ${OUTPUT_PATH}..." mkdir -p "${OUTPUT_PATH}" - cp -r result_dir/. "${OUTPUT_PATH}/" + cp -r "${RUN_RESULT_DIR}/." "${OUTPUT_PATH}/" echo "Results copied to ${OUTPUT_PATH}" fi + +[[ "${JOB_STATE}" == "COMPLETED" ]] || exit 1 diff --git a/presto/slurm/presto-nvl72/launcher_common.sh b/presto/slurm/presto-nvl72/launcher_common.sh index a05f1783..a2df6fc8 100644 --- a/presto/slurm/presto-nvl72/launcher_common.sh +++ b/presto/slurm/presto-nvl72/launcher_common.sh @@ -13,7 +13,12 @@ # resolve_cluster_variant # Reads CLUSTER_{GPU,CPU}_* and populates the generic CLUSTER_DEFAULT_*, # CLUSTER_CPUS_PER_TASK, CLUSTER_NUM_WORKERS_PER_NODE, CLUSTER_TIME_*, -# CLUSTER_DEFAULT_PORT, CLUSTER_UCX_NET_DEVICES, CLUSTER_USE_NUMA, +# CLUSTER_DEFAULT_PORT, CLUSTER_UCX_NET_DEVICES, +# CLUSTER_UCX_NET_DEVICES_BY_LOCAL_WORKER, +# CLUSTER_UCX_NET_DEVICES_BY_GPU, +# CLUSTER_INTERNAL_ADDRESS_INTERFACE, +# CLUSTER_INTERNAL_ADDRESS_INTERFACE_BY_GPU, +# CLUSTER_MELLANOX_VISIBLE_DEVICES, CLUSTER_USE_NUMA, # CLUSTER_EXTRA_MOUNTS, CLUSTER_NUMA_GPUS_PER_NODE, CLUSTER_LIB*_PATH # variables, plus COORD_IMAGE / WORKER_IMAGE. Pre-existing values are # preserved (so command-line flags and shell exports still win). @@ -57,11 +62,31 @@ _path_is_compute_only() { local path="$1" prefix [[ -z "${CLUSTER_COMPUTE_ONLY_PATHS:-}" ]] && return 1 for prefix in ${CLUSTER_COMPUTE_ONLY_PATHS}; do - [[ -n "${prefix}" && "${path}" == "${prefix}"* ]] && return 0 + # Match the prefix itself or a path below it, but not a sibling such as + # /scratch-old when the configured prefix is /scratch. + prefix="${prefix%/}" + [[ -n "${prefix}" && ( "${path}" == "${prefix}" || "${path}" == "${prefix}/"* ) ]] && return 0 done return 1 } +# canonicalize_file_path +# Print an absolute path. Existing paths are normalized through their physical +# parent directory; compute-only paths that are invisible on the login node are +# made absolute without requiring that parent to exist locally. +canonicalize_file_path() { + local path="$1" parent base + if [[ "${path}" != /* ]]; then + path="${PWD}/${path}" + fi + parent="$(dirname -- "${path}")" + base="$(basename -- "${path}")" + if [[ -d "${parent}" ]]; then + parent="$(cd -- "${parent}" && pwd -P)" + fi + printf '%s/%s\n' "${parent%/}" "${base}" +} + # resolve_image_path # Resolve a container image name or path to an absolute .sqsh path. # Accepts: bare name (appends .sqsh under IMAGE_DIR), name ending in .sqsh @@ -95,6 +120,22 @@ preflight_image() { fi } +# preflight_image_roles +# Catch the common -w/-c transposition before consuming a Slurm allocation. +# This is deliberately a narrow name-based check: generated images in this +# repository contain "coordinator" only in the Java coordinator artifact. +preflight_image_roles() { + local worker_image="$1" coord_image="$2" + local worker_lower="${worker_image,,}" coord_lower="${coord_image,,}" + if [[ "${worker_lower}" == *coordinator* && "${coord_lower}" != *coordinator* ]]; then + echo "Error: worker and coordinator images appear to be reversed." >&2 + echo " -w/--worker-image: ${worker_image}" >&2 + echo " -c/--coord-image: ${coord_image}" >&2 + echo " Pass the *-coordinator-* image to -c and the *-cpu-* or *-gpu-* image to -w." >&2 + exit 1 + fi +} + # preflight_dir [] # Verify a directory exists. is used in the error message. preflight_dir() { @@ -115,6 +156,26 @@ preflight_dir() { fi } +# preflight_file [] +# Verify a regular file exists. is used in the error message. +preflight_file() { + local path="$1" desc="$2" hint="${3:-}" + if [[ -z "${path}" ]]; then + echo "Error: ${desc} path is not set" >&2 + [[ -n "${hint}" ]] && echo " To fix: ${hint}" >&2 + exit 1 + fi + if _path_is_compute_only "${path}"; then + echo "Note: skipping host-side preflight for ${desc} at ${path} (compute-only path)" >&2 + return 0 + fi + if [[ ! -f "${path}" ]]; then + echo "Error: ${desc} file not found at ${path}" >&2 + [[ -n "${hint}" ]] && echo " To fix: ${hint}" >&2 + exit 1 + fi +} + # ---------------------------------------------------------------------------- # Job monitoring / output # ---------------------------------------------------------------------------- @@ -278,6 +339,11 @@ resolve_cluster_variant() { _resolve_var CLUSTER_TIME_ANALYZE "${prefix}_TIME_ANALYZE" _resolve_var CLUSTER_DEFAULT_PORT "${prefix}_DEFAULT_PORT" _resolve_var CLUSTER_UCX_NET_DEVICES "${prefix}_UCX_NET_DEVICES" + _resolve_var CLUSTER_UCX_NET_DEVICES_BY_LOCAL_WORKER "${prefix}_UCX_NET_DEVICES_BY_LOCAL_WORKER" + _resolve_var CLUSTER_UCX_NET_DEVICES_BY_GPU "${prefix}_UCX_NET_DEVICES_BY_GPU" + _resolve_var CLUSTER_INTERNAL_ADDRESS_INTERFACE "${prefix}_INTERNAL_ADDRESS_INTERFACE" + _resolve_var CLUSTER_INTERNAL_ADDRESS_INTERFACE_BY_GPU "${prefix}_INTERNAL_ADDRESS_INTERFACE_BY_GPU" + _resolve_var CLUSTER_MELLANOX_VISIBLE_DEVICES "${prefix}_MELLANOX_VISIBLE_DEVICES" _resolve_var CLUSTER_EXTRA_MOUNTS "${prefix}_EXTRA_MOUNTS" _resolve_var COORD_IMAGE "${prefix}_DEFAULT_COORD_IMAGE" _resolve_var WORKER_IMAGE "${prefix}_DEFAULT_WORKER_IMAGE" @@ -309,20 +375,26 @@ build_common_export_vars() { EXPORT_VARS+=",WORKER_ENV_FILE=${WORKER_ENV_FILE}" EXPORT_VARS+=",CLUSTER_DEFAULT_PORT=${CLUSTER_DEFAULT_PORT}" local v - for v in CLUSTER_UCX_NET_DEVICES CLUSTER_NUMA_GPUS_PER_NODE \ + for v in CLUSTER_INTERNAL_ADDRESS_INTERFACE \ + CLUSTER_MELLANOX_VISIBLE_DEVICES CLUSTER_NUMA_GPUS_PER_NODE \ CLUSTER_LIBCUDA_HOST_PATH CLUSTER_LIBCUDA_CONTAINER_PATH \ CLUSTER_LIBNVIDIA_ML_HOST_PATH CLUSTER_LIBNVIDIA_ML_CONTAINER_PATH \ - CLUSTER_EXTRA_MOUNTS CLUSTER_CONFIG \ + CLUSTER_CONFIG \ HIVE_METASTORE_VERSION HIVE_METASTORE_SHARED_ROOT; do - [[ -n "${!v:-}" ]] || continue - if [[ "${!v}" == *,* ]]; then - # Values with commas can't be inlined into --export (comma is the - # separator); export to the environment so --export=ALL picks them up. - export "${v}=${!v}" - else - EXPORT_VARS+=",${v}=${!v}" - fi + [[ -n "${!v:-}" ]] && EXPORT_VARS+=",${v}=${!v}" + done + # Values in these variables may contain commas (rail lists/mount lists), so + # they cannot be embedded directly in sbatch's comma-delimited --export + # argument. Export them into the caller and let the leading ALL carry each + # value without re-tokenizing it. + for v in CLUSTER_UCX_NET_DEVICES \ + CLUSTER_UCX_NET_DEVICES_BY_LOCAL_WORKER \ + CLUSTER_UCX_NET_DEVICES_BY_GPU CLUSTER_EXTRA_MOUNTS \ + CLUSTER_INTERNAL_ADDRESS_INTERFACE_BY_GPU \ + PRESTO_INTERNAL_ADDRESS_INTERFACE_BY_GPU; do + [[ -n "${!v:-}" ]] && export "${v}" done + return 0 } build_cluster_sbatch_args() { diff --git a/presto/slurm/presto-nvl72/numa-worker-launch.sh b/presto/slurm/presto-nvl72/numa-worker-launch.sh new file mode 100644 index 00000000..9a035fbe --- /dev/null +++ b/presto/slurm/presto-nvl72/numa-worker-launch.sh @@ -0,0 +1,33 @@ +#!/bin/bash +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +set -euo pipefail + +if (( $# < 2 )); then + echo "Usage: $0 [args...]" >&2 + exit 2 +fi + +cpu_numa_node=$1 +shift +if [[ ! "${cpu_numa_node}" =~ ^[0-9]+$ ]]; then + echo "Invalid CPU NUMA node: ${cpu_numa_node}" >&2 + exit 2 +fi + +cpu_bind_option="cpunodebind" +echo "Requested worker NUMA policy: --${cpu_bind_option}=${cpu_numa_node} --membind=${cpu_numa_node}" + +# Enter the requested CPU and memory policies before inspecting them. exec +# preserves the PID across this diagnostic shell and the final Presto server, +# which is required by the host-side perf attachment path. +exec numactl \ + "--${cpu_bind_option}=${cpu_numa_node}" \ + --membind="${cpu_numa_node}" \ + /bin/bash -c ' + echo "Effective worker NUMA policy:" + numactl --show 2>&1 || true + grep -E "^(Cpus_allowed_list|Mems_allowed_list):" /proc/self/status || true + exec "$@" + ' _ "$@" diff --git a/presto/slurm/presto-nvl72/run-presto-benchmarks.sh b/presto/slurm/presto-nvl72/run-presto-benchmarks.sh index bbb22135..8527585b 100755 --- a/presto/slurm/presto-nvl72/run-presto-benchmarks.sh +++ b/presto/slurm/presto-nvl72/run-presto-benchmarks.sh @@ -14,10 +14,51 @@ set -exuo pipefail source $SCRIPT_DIR/echo_helpers.sh source $SCRIPT_DIR/functions.sh -# Ensure metadata injection runs even if the script exits early (e.g. a worker -# fails to register). This guarantees benchmark_result.json always has a -# context block with image_digest before the results are copied out. -trap 'inject_benchmark_metadata' EXIT +function collect_coordinator_queries { + if [[ -x "${SCRIPT_DIR}/collect-coordinator-queries.sh" && -n "${COORD:-}" && -n "${PORT:-}" ]]; then + "${SCRIPT_DIR}/collect-coordinator-queries.sh" \ + --server "http://${COORD}:${PORT}" \ + --output-dir "${RESULT_DIR:-${SCRIPT_DIR}/result_dir}/coordinator_queries" || true + fi +} + +# Ensure metadata/log collection runs even if the script exits early (e.g. a +# query OOMs). DEBUG_HOLD_ON_FAILURE_SECONDS keeps the Slurm allocation alive +# long enough to inspect the coordinator and worker logs before teardown. +function on_exit { + local rc=$? + trap - EXIT + + # A perf capture may still be active when pytest or verification fails. + # Stop it and publish every node-local raw capture before Slurm tears down + # worker 0. Optional report generation also runs here, after all queries. + stop_perf_sampler || true + inject_benchmark_metadata || true + + if [[ "${rc}" != "0" ]]; then + collect_coordinator_queries + echo "Benchmark failed with exit code ${rc}; collecting logs before teardown." + collect_results || true + + if [[ -n "${DEBUG_HOLD_ON_FAILURE_SECONDS:-}" && "${DEBUG_HOLD_ON_FAILURE_SECONDS}" != "0" ]]; then + echo "Holding failed cluster for ${DEBUG_HOLD_ON_FAILURE_SECONDS}s." + echo "Coordinator: http://${COORD}:${PORT}" + echo "Logs: ${LOGS}" + echo "Result dir: ${RESULT_DIR:-${SCRIPT_DIR}/result_dir}" + sleep "${DEBUG_HOLD_ON_FAILURE_SECONDS}" || true + fi + fi + + if ! stop_cluster; then + echo "Cluster cleanup did not complete cleanly." >&2 + [[ "${rc}" == "0" ]] && rc=1 + fi + # Capture teardown and port-release diagnostics as well as startup logs. + collect_results || true + + exit "${rc}" +} +trap on_exit EXIT # ============================================================================== # Setup and Validation @@ -35,21 +76,31 @@ start_cluster # ============================================================================== echo "Running TPC-H queries (${NUM_ITERATIONS} iterations, scale factor ${SCALE_FACTOR})..." run_queries ${NUM_ITERATIONS} ${SCALE_FACTOR} +# Host perf must publish every node-local raw capture before worker teardown. +# Optional post-processing happens only now, after all measured queries. This is +# a no-op for ordinary and nsys runs. +stop_perf_sampler +collect_coordinator_queries +verify_cpu_ucx_runtime # ============================================================================== # Process Results # ============================================================================== echo "Processing results..." -mkdir -p ${SCRIPT_DIR}/result_dir -cp -r ${LOGS}/cli.log ${SCRIPT_DIR}/result_dir/summary.txt +mkdir -p "${RESULT_DIR:-${SCRIPT_DIR}/result_dir}" +cp -r "${LOGS}/cli.log" "${RESULT_DIR:-${SCRIPT_DIR}/result_dir}/summary.txt" echo "Collecting configs and logs into result directory..." collect_results wait_for_nsys_report_generation +stop_cluster +# Include explicit shutdown and port-release diagnostics in the immutable +# job-specific result directory. +collect_results echo "========================================" echo "Benchmark complete!" -echo "Results saved to: ${SCRIPT_DIR}/results_dir" +echo "Results saved to: ${RESULT_DIR:-${SCRIPT_DIR}/result_dir}" echo "Logs available at: ${LOGS}" echo "========================================" diff --git a/presto/slurm/presto-nvl72/slurm_common.sh b/presto/slurm/presto-nvl72/slurm_common.sh index 3598f92b..8301ed0b 100644 --- a/presto/slurm/presto-nvl72/slurm_common.sh +++ b/presto/slurm/presto-nvl72/slurm_common.sh @@ -27,6 +27,7 @@ export VT_ROOT source "${SCRIPT_DIR}/defaults.env" export DATA IMAGE_DIR HIVE_METASTORE_SHARED_ROOT HIVE_METASTORE_VERSION export LOGS="${SCRIPT_DIR}/logs" +export RESULT_DIR="${RESULT_DIR:-${SCRIPT_DIR}/result_dir}" export VARIANT_TYPE="${VARIANT_TYPE:-gpu}" export USE_NUMA="${USE_NUMA:-0}" export CONFIGS="${VT_ROOT}/presto/docker/config/generated/${VARIANT_TYPE}" @@ -36,19 +37,44 @@ export NUM_NODES="${SLURM_JOB_NUM_NODES}" COORD="$(scontrol show hostnames "${SLURM_JOB_NODELIST}" | head -n 1)" export COORD export NUM_WORKERS=$((NUM_NODES * NUM_GPUS_PER_NODE)) +# generate_presto_config.sh runs once on a host but describes the full cluster. +# Keep local resource partitioning separate from NUM_WORKERS so a two-host, +# one-worker-per-host CPU run does not mistakenly divide each host by two. +export CPU_WORKERS_PER_HOST="${CPU_WORKERS_PER_HOST:-${NUM_GPUS_PER_NODE}}" # Presto configuration export PORT="${PORT:-${CLUSTER_DEFAULT_PORT:?CLUSTER_DEFAULT_PORT not set — check cluster_config.env}}" export CUDF_LIB=/usr/lib64/presto-native-libs -# UCX configuration (common subset; .slurm files may add mode-specific knobs) -export UCX_TLS=^ib,ud:aux,sm -if [[ -n "${CLUSTER_UCX_NET_DEVICES:-}" ]]; then +# UCX configuration (common subset; .slurm files may add mode-specific knobs). +# Keep exported UCX_* values when the launcher sets an experiment explicitly. +# Remapped CPU containers default to sysv for same-host traffic and TCP between +# hosts; explicit experiments can still select POSIX or RDMA transports. +# Retain the existing conservative GPU default for runs that do not opt in. +_ucx_tls_default="^ib,ud:aux,sm" +[[ "${VARIANT_TYPE}" == "cpu" ]] && _ucx_tls_default="sysv,tcp,self" +export UCX_TLS="${UCX_TLS:-${_ucx_tls_default}}" +unset _ucx_tls_default +if [[ -n "${UCX_NET_DEVICES:-}" ]]; then + export UCX_NET_DEVICES +elif [[ -n "${CLUSTER_UCX_NET_DEVICES:-}" ]]; then export UCX_NET_DEVICES="${CLUSTER_UCX_NET_DEVICES}" fi -export UCX_RNDV_PIPELINE_ERROR_HANDLING=y -export UCX_TCP_KEEPINTVL=1ms -export UCX_KEEPALIVE_INTERVAL=1ms +if [[ "${UCX_TLS}" == *^ib* && ( "${UCX_NET_DEVICES:-}" == *mlx5_* || "${CLUSTER_UCX_NET_DEVICES_BY_LOCAL_WORKER:-}" == *mlx5_* || "${CLUSTER_UCX_NET_DEVICES_BY_GPU:-}" == *mlx5_* ) ]]; then + echo "Error: the configured UCX device selection includes mlx5 RDMA, but UCX_TLS=${UCX_TLS} excludes IB/RDMA." >&2 + echo "Set UCX_TLS explicitly, e.g. UCX_TLS=rc,cuda_copy,cuda_ipc,self, or use a TCP netdev such as bond0." >&2 + exit 1 +fi +if [[ -n "${CLUSTER_UCX_NET_DEVICES_BY_GPU:-}" && -z "${PRESTO_INTERNAL_ADDRESS_INTERFACE_BY_GPU:-${CLUSTER_INTERNAL_ADDRESS_INTERFACE_BY_GPU:-}}" ]]; then + echo "Warning: CLUSTER_UCX_NET_DEVICES_BY_GPU is set without a per-GPU internal address interface map." >&2 + echo " Each worker must advertise an IP on the same rail as its UCX_NET_DEVICES entry." >&2 +fi +export UCX_RNDV_PIPELINE_ERROR_HANDLING="${UCX_RNDV_PIPELINE_ERROR_HANDLING:-y}" +export UCX_TCP_KEEPINTVL="${UCX_TCP_KEEPINTVL:-1ms}" +export UCX_KEEPALIVE_INTERVAL="${UCX_KEEPALIVE_INTERVAL:-1ms}" +[[ -n "${UCX_MAX_RNDV_RAILS:-}" ]] && export UCX_MAX_RNDV_RAILS +[[ -n "${UCX_PROTO_INFO:-}" ]] && export UCX_PROTO_INFO +[[ -n "${UCX_LOG_LEVEL:-}" ]] && export UCX_LOG_LEVEL mkdir -p "${LOGS}" @@ -66,4 +92,6 @@ echo "Variant: ${VARIANT_TYPE}" echo "Data directory: ${DATA}" echo "Image directory: ${IMAGE_DIR}" echo "Logs directory: ${LOGS}" +echo "Results directory: ${RESULT_DIR}" echo "Total workers: ${NUM_WORKERS} (${NUM_NODES} nodes x ${NUM_GPUS_PER_NODE} workers/node)" +[[ "${VARIANT_TYPE}" == "cpu" ]] && echo "CPU workers per host: ${CPU_WORKERS_PER_HOST}" diff --git a/presto/slurm/presto-nvl72/worker.env b/presto/slurm/presto-nvl72/worker.env index c62e68c5..843d6e22 100644 --- a/presto/slurm/presto-nvl72/worker.env +++ b/presto/slurm/presto-nvl72/worker.env @@ -2,3 +2,28 @@ KVIKIO_TASK_SIZE=$((16*1024*1024)) KVIKIO_NTHREADS=16 +UCX_LOG_LEVEL="${UCX_LOG_LEVEL:-warn}" +UCX_PROTO_INFO="${UCX_PROTO_INFO:-n}" +UCX_WARN_UNUSED_ENV_VARS="${UCX_WARN_UNUSED_ENV_VARS:-n}" + +# CPU-row UCX exchange runtime defaults. This file is sourced with `set -a`, so +# assignments become process environment variables. Keep these CPU-only: the +# same worker.env is also used by GPU launches. Set VELOX_UCX_CPU_EXCHANGE=0 +# in the shell or this file when an HTTP-exchange CPU baseline is wanted. +if [[ "${VARIANT_TYPE:-gpu}" == "cpu" ]]; then + VELOX_UCX_CPU_EXCHANGE="${VELOX_UCX_CPU_EXCHANGE:-1}" + + # CPU UCX development images install the matching UCX runtime and modules + # under /usr/local. Prefer those over any older UCX copy bundled in + # /usr/lib64/presto-native-libs. Both settings remain shell-overridable. + PRESTO_WORKER_UCX_LOCAL_LIB_FIRST="${PRESTO_WORKER_UCX_LOCAL_LIB_FIRST:-1}" + UCX_MODULE_DIR="${UCX_MODULE_DIR:-/usr/local/lib/ucx}" + + # functions.sh derives VELOX_UCX_CPU_PORT independently for every worker + # from that worker's generated http-server.http.port (HTTP+3). Avoid POSIX + # by default because remapped workers cannot open peer /proc//fd paths. + UCX_TLS="${UCX_TLS:-sysv,tcp,self}" + UCX_NET_DEVICES="${UCX_NET_DEVICES:-all}" + UCX_POSIX_USE_PROC_LINK="${UCX_POSIX_USE_PROC_LINK:-n}" + +fi