[BugFix] stage Phase 2 communication tool on workers

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

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"