diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh index 034225e..5f641e0 100755 --- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh @@ -12,6 +12,9 @@ RUN_ID="${RUN_ID:-dsv4pro-phase2-$(date +%Y%m%d-%H%M%S)}" RESULT_DIR="${RESULT_BASE}/${RUN_ID}" RESULT_TOOL="${SCRIPT_DIR}/hardware_contention_attribution.py" COMMUNICATION_TOOL="${SCRIPT_DIR}/communication_baseline.py" +COMMUNICATION_STAGE_DIR="/tmp/${EXPERIMENT}_communication/${RUN_ID}" +COMMUNICATION_STAGE_TOOL="${COMMUNICATION_STAGE_DIR}/communication_baseline.py" +COMMUNICATION_CONTAINER_TOOL="/opt/phase2/communication_baseline.py" MARKERS_PATH="${RESULT_DIR}/markers.csv" SERVICE_DIR="${RESULT_DIR}/service" COMMAND_DIR="${RESULT_DIR}/commands" @@ -32,6 +35,15 @@ print_command() { printf '\n' } +sha256_file() { + local path="$1" + if command -v sha256sum >/dev/null 2>&1; then + sha256sum "${path}" | awk '{print $1}' + else + shasum -a 256 "${path}" | awk '{print $1}' + fi +} + node_is_local() { local node="$1" local local_ips @@ -133,7 +145,7 @@ validate_config() { preflight_node_tools() { local node="$1" - local required="nvidia-smi docker mpstat pidstat sar numastat perf" + local required="nvidia-smi docker mpstat pidstat sar numastat perf sha256sum" local tool for tool in ${required}; do if ! run_on_node "${node}" "command -v '${tool}' >/dev/null"; then @@ -266,6 +278,46 @@ run_phase1_action() { "${command[@]}" } +stage_communication_tool() { + local provenance_tool="${RESULT_DIR}/communication/communication_baseline.py" + local provenance_hash="${RESULT_DIR}/communication/communication_baseline.sha256" + mkdir -p "${RESULT_DIR}/communication" + install -m 0644 "${COMMUNICATION_TOOL}" "${provenance_tool}" + printf '%s %s\n' \ + "$(sha256_file "${provenance_tool}")" \ + "$(basename "${provenance_tool}")" \ + > "${provenance_hash}" + + if [[ "${DRY_RUN}" == "1" ]]; then + log "[DRY] stage communication tool on both nodes: ${COMMUNICATION_STAGE_TOOL}" + return 0 + fi + + local expected_hash actual_hash node + expected_hash="$(sha256_file "${COMMUNICATION_TOOL}")" + for node in "${HEAD_NODE}" "${WORKER_NODE}"; do + if node_is_local "${node}"; then + mkdir -p "${COMMUNICATION_STAGE_DIR}" + install -m 0644 "${COMMUNICATION_TOOL}" "${COMMUNICATION_STAGE_TOOL}" + else + ssh -o BatchMode=yes -o StrictHostKeyChecking=no "${node}" \ + "mkdir -p '${COMMUNICATION_STAGE_DIR}' && \ + cat > '${COMMUNICATION_STAGE_TOOL}' && \ + chmod 0644 '${COMMUNICATION_STAGE_TOOL}'" \ + < "${COMMUNICATION_TOOL}" + fi + actual_hash="$( + run_on_node "${node}" "sha256sum '${COMMUNICATION_STAGE_TOOL}'" | + cut -d' ' -f1 + )" + if [[ "${actual_hash}" != "${expected_hash}" ]]; then + log "ERROR: staged communication tool hash mismatch node=${node}" + return 1 + fi + log "staged communication tool node=${node} sha256=${actual_hash}" + done +} + build_communication_docker_command() { local output_name="$1" local container_name="$2" @@ -281,7 +333,7 @@ build_communication_docker_command() { --shm-size 20g --ulimit memlock=-1 --ulimit stack=67108864 - -v "${REPO_ROOT}:${REPO_ROOT}:ro" + -v "${COMMUNICATION_STAGE_TOOL}:${COMMUNICATION_CONTAINER_TOOL}:ro" -e "NCCL_SOCKET_IFNAME=${NCCL_SOCKET_IFNAME}" -e "NCCL_IB_HCA=${NCCL_IB_HCA}" -e "NCCL_CROSS_NIC=${cross_nic}" @@ -304,7 +356,7 @@ run_p2p_baseline() { local command build_communication_docker_command \ command "${container}" python3 2 \ - "${COMMUNICATION_TOOL}" p2p \ + "${COMMUNICATION_CONTAINER_TOOL}" p2p \ --node "${role}" \ --size "${P2P_SIZE}" \ --warmup "${P2P_WARMUP}" \ @@ -333,7 +385,7 @@ run_single_node_allreduce() { --nnodes=1 \ --nproc-per-node=8 \ --master-port "${master_port}" \ - "${COMMUNICATION_TOOL}" all-reduce \ + "${COMMUNICATION_CONTAINER_TOOL}" all-reduce \ --scope "${role}_8gpu" \ --sizes "${COMMUNICATION_SIZES}" \ --repetitions "${COMMUNICATION_REPETITIONS}" \ @@ -367,7 +419,7 @@ run_two_node_allreduce() { worker_command "${worker_container}" torchrun "${cross_nic}" \ "${common_args[@]}" \ --node-rank=1 \ - "${COMMUNICATION_TOOL}" all-reduce \ + "${COMMUNICATION_CONTAINER_TOOL}" all-reduce \ --scope two_node_16gpu \ --sizes "${COMMUNICATION_SIZES}" \ --repetitions "${COMMUNICATION_REPETITIONS}" \ @@ -377,7 +429,7 @@ run_two_node_allreduce() { head_command "${head_container}" torchrun "${cross_nic}" \ "${common_args[@]}" \ --node-rank=0 \ - "${COMMUNICATION_TOOL}" all-reduce \ + "${COMMUNICATION_CONTAINER_TOOL}" all-reduce \ --scope two_node_16gpu \ --sizes "${COMMUNICATION_SIZES}" \ --repetitions "${COMMUNICATION_REPETITIONS}" \ @@ -424,6 +476,19 @@ cleanup_communication_containers() { done } +cleanup_communication_runtime() { + [[ "${DRY_RUN}" == "1" ]] && return 0 + local node + for node in "${HEAD_NODE}" "${WORKER_NODE}"; do + run_on_node "${node}" "rm -rf '${COMMUNICATION_STAGE_DIR}'" || true + done +} + +cleanup_communication() { + cleanup_communication_containers + cleanup_communication_runtime +} + run_communication_baseline() { [[ "${RUN_COMMUNICATION_BASELINE}" == "1" ]] || { log "SKIP communication baseline by configuration" @@ -434,6 +499,7 @@ run_communication_baseline() { preflight_gpus_idle "${HEAD_NODE}" preflight_gpus_idle "${WORKER_NODE}" fi + stage_communication_tool run_p2p_baseline head "${HEAD_NODE}" run_p2p_baseline worker "${WORKER_NODE}" run_single_node_allreduce head "${HEAD_NODE}" "${COMMUNICATION_MASTER_PORT}" @@ -443,6 +509,7 @@ run_communication_baseline() { run_two_node_allreduce "${cross_nic}" done cleanup_communication_containers + cleanup_communication_runtime } start_service() { @@ -850,6 +917,7 @@ cleanup() { stop_collectors || true stop_service || true cleanup_communication_containers || true + cleanup_communication_runtime || true if (( rc != 0 )) && [[ -f "${RESULT_DIR}/manifest.json" ]]; then finish_manifest ABORTED || true fi @@ -935,7 +1003,7 @@ run_communication_only() { preflight_node_tools "${WORKER_NODE}" preflight_clock_sync fi - trap cleanup_communication_containers EXIT INT TERM + trap cleanup_communication EXIT INT TERM run_communication_baseline python3 "${RESULT_TOOL}" summarize-communication "${RESULT_DIR}" trap - EXIT INT TERM diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py index 66df919..d46bbdf 100644 --- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py @@ -15,6 +15,20 @@ import hardware_contention_attribution as attribution # noqa: E402 class HardwareContentionAttributionTest(unittest.TestCase): + def test_communication_tool_is_staged_without_worker_repo_dependency( + self, + ) -> None: + script = ( + EXPERIMENT_DIR / "run_hardware_contention_attribution.sh" + ).read_text(encoding="utf-8") + + self.assertIn("stage_communication_tool", script) + self.assertIn( + '-v "${COMMUNICATION_STAGE_TOOL}:${COMMUNICATION_CONTAINER_TOOL}:ro"', + script, + ) + self.assertNotIn('-v "${REPO_ROOT}:${REPO_ROOT}:ro"', script) + def test_gpu_summary_ignores_na_and_computes_percentiles(self) -> None: with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "gpu_samples.csv"